Skip to main content

max / makenotwork

6.0 KB · 205 lines History Blame Raw
1 //! The front door.
2 //!
3 //! [`Chat`] owns the crate's mutable state (the fan-out hub and the send
4 //! budgets) so a host holds one thing in application state rather than
5 //! assembling three and hoping it wires them together the same way twice.
6
7 use std::time::{Duration, Instant};
8
9 use crate::error::ChatError;
10 use crate::hub::{Hub, HubLimits};
11 use crate::ids::{MessageId, Nonce, RoomId, UserId};
12 use crate::message::Message;
13 use crate::rate_limit::{RateLimiter, RateLimits};
14 use crate::room::Room;
15 use crate::stream::ChatStream;
16 use crate::traits::ChatAuthz;
17
18 /// One send attempt.
19 #[derive(Debug, Clone)]
20 pub struct SendRequest<'a> {
21 pub room: &'a Room,
22 pub author: UserId,
23 /// Unvalidated. Trimmed and length-checked on the way through.
24 pub body: &'a str,
25 /// The sender's optimistic-render nonce, echoed back so their own message
26 /// reconciles instead of appearing twice.
27 pub nonce: Option<Nonce>,
28 /// Injected so the send path is testable without sleeping.
29 pub now: Instant,
30 }
31
32 /// Chat state for one process.
33 ///
34 /// Cheap to clone: the hub is an `Arc` internally and the limiter is shared, so
35 /// clones talk to the same rooms and the same budgets.
36 pub struct Chat {
37 hub: Hub,
38 limiter: RateLimiter,
39 }
40
41 impl Chat {
42 pub fn new(hub_limits: HubLimits, rate_limits: RateLimits) -> Self {
43 Self {
44 hub: Hub::new(hub_limits),
45 limiter: RateLimiter::new(rate_limits),
46 }
47 }
48
49 pub fn hub(&self) -> &Hub {
50 &self.hub
51 }
52
53 pub fn limiter(&self) -> &RateLimiter {
54 &self.limiter
55 }
56
57 /// Validate, authorize, store, and fan out one message.
58 ///
59 /// See [`crate::send`] for why the order of those steps is a security
60 /// property rather than a preference.
61 pub async fn send<A, F, Fut>(
62 &self,
63 authz: &A,
64 req: SendRequest<'_>,
65 store: F,
66 ) -> Result<Message, ChatError>
67 where
68 A: ChatAuthz + ?Sized,
69 F: FnOnce(crate::message::MessageBody) -> Fut,
70 Fut: Future<Output = Result<Message, ChatError>>,
71 {
72 crate::send::send(&self.hub, &self.limiter, authz, req, store).await
73 }
74
75 /// Open a stream: backlog from `after`, then the live feed.
76 ///
77 /// Subscribes before fetching. See [`ChatStream`] for why that order is the
78 /// whole correctness question.
79 pub async fn subscribe<F, Fut>(
80 &self,
81 room: RoomId,
82 user: UserId,
83 after: Option<MessageId>,
84 fetch: F,
85 ) -> Result<ChatStream, ChatError>
86 where
87 F: FnOnce(Option<MessageId>) -> Fut,
88 Fut: Future<Output = Result<Vec<Message>, ChatError>>,
89 {
90 ChatStream::open(&self.hub, room, user, after, fetch).await
91 }
92
93 /// Drop rate-limit buckets nobody has touched recently.
94 ///
95 /// The host must call this on a timer. Bucket state is keyed by (user, room),
96 /// which is unbounded and attacker-influenced, and both consumers run under a
97 /// hard 512M cgroup cap. Uses the shortest interval that cannot turn eviction
98 /// into a way to skip the queue.
99 ///
100 /// Returns how many buckets were dropped.
101 pub fn sweep(&self, now: Instant) -> usize {
102 self.limiter
103 .evict_idle(now, self.limiter.recommended_idle_timeout())
104 }
105
106 /// How often [`Chat::sweep`] wants to be called.
107 pub fn sweep_interval(&self) -> Duration {
108 self.limiter.recommended_idle_timeout()
109 }
110 }
111
112 #[cfg(test)]
113 mod tests {
114 use super::*;
115 use crate::retention::Retention;
116 use crate::room::RoomState;
117 use crate::traits::WriteAccess;
118 use async_trait::async_trait;
119 use uuid::Uuid;
120
121 struct Yes;
122
123 #[async_trait]
124 impl ChatAuthz for Yes {
125 async fn can_read(&self, _v: Option<UserId>, _r: &Room) -> Result<bool, ChatError> {
126 Ok(true)
127 }
128 async fn can_write(&self, _u: UserId, _r: &Room) -> Result<WriteAccess, ChatError> {
129 Ok(WriteAccess::Allow)
130 }
131 async fn is_moderator(&self, _u: UserId, _r: &Room) -> Result<bool, ChatError> {
132 Ok(false)
133 }
134 }
135
136 fn room() -> Room {
137 Room {
138 id: RoomId(Uuid::new_v4()),
139 state: RoomState::Open,
140 retention: Retention::forum_default(),
141 }
142 }
143
144 #[tokio::test]
145 async fn send_and_subscribe_share_the_same_rooms() {
146 let chat = Chat::new(HubLimits::default(), RateLimits::default());
147 let r = room();
148 let author = UserId(Uuid::new_v4());
149
150 let mut stream = chat
151 .subscribe(r.id, UserId(Uuid::new_v4()), None, |_| async { Ok(vec![]) })
152 .await
153 .unwrap();
154
155 chat.send(
156 &Yes,
157 SendRequest {
158 room: &r,
159 author,
160 body: "hi",
161 nonce: None,
162 now: Instant::now(),
163 },
164 |body| async move {
165 Ok(Message {
166 id: MessageId(1),
167 room_id: r.id,
168 author_id: author,
169 body_html: body.as_str().to_owned(),
170 created_at: 0,
171 nonce: None,
172 })
173 },
174 )
175 .await
176 .unwrap();
177
178 assert!(
179 matches!(
180 stream.next().await,
181 Some(crate::event::ChatEvent::Message(_))
182 ),
183 "a message sent through Chat reaches a stream opened through Chat"
184 );
185 }
186
187 #[test]
188 fn sweep_uses_an_interval_that_is_not_a_bypass() {
189 let chat = Chat::new(
190 HubLimits::default(),
191 RateLimits {
192 burst: 5,
193 sustain_per_min: 60,
194 },
195 );
196 assert_eq!(chat.sweep_interval(), Duration::from_secs(5));
197
198 let t0 = Instant::now();
199 chat.limiter()
200 .check(UserId(Uuid::nil()), RoomId(Uuid::nil()), t0);
201 assert_eq!(chat.sweep(t0), 0, "a fresh bucket is not idle");
202 assert_eq!(chat.sweep(t0 + Duration::from_secs(10)), 1);
203 }
204 }
205