Skip to main content

max / makenotwork

8.1 KB · 275 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 /// Remove one message and tell the room.
94 ///
95 /// `actor` is the author for a self-delete or a moderator otherwise; the
96 /// host's impl proves the entitlement, since only it can read the row.
97 pub async fn delete_message<M>(
98 &self,
99 moderation: &M,
100 actor: UserId,
101 room: &Room,
102 message: MessageId,
103 ) -> Result<(), ChatError>
104 where
105 M: crate::traits::ChatModeration + ?Sized,
106 {
107 crate::moderate::delete_message(&self.hub, moderation, actor, room, message).await
108 }
109
110 /// Remove every message by `target` in the room, announced as one event.
111 ///
112 /// Returns how many rows went. See [`crate::moderate`] for why this is not
113 /// N deletes.
114 pub async fn purge_user<M>(
115 &self,
116 moderation: &M,
117 actor: UserId,
118 room: &Room,
119 target: UserId,
120 ) -> Result<u64, ChatError>
121 where
122 M: crate::traits::ChatModeration + ?Sized,
123 {
124 crate::moderate::purge_user(&self.hub, moderation, actor, room, target).await
125 }
126
127 /// Ban `target` and remove their backlog, returning how many messages went.
128 ///
129 /// The pairing is here rather than left to the caller because a ban that
130 /// forgets the purge leaves the damage up, and that is the whole reason the
131 /// two go together.
132 pub async fn ban_user<M>(
133 &self,
134 moderation: &M,
135 actor: UserId,
136 room: &Room,
137 target: UserId,
138 reason: Option<&str>,
139 ) -> Result<u64, ChatError>
140 where
141 M: crate::traits::ChatModeration + ?Sized,
142 {
143 crate::moderate::ban_user(&self.hub, moderation, actor, room, target, reason).await
144 }
145
146 /// Stop `target` from sending for `duration`. Reads are unaffected and the
147 /// room is told nothing.
148 pub async fn timeout_user<M>(
149 &self,
150 moderation: &M,
151 actor: UserId,
152 room: &Room,
153 target: UserId,
154 duration: Duration,
155 ) -> Result<(), ChatError>
156 where
157 M: crate::traits::ChatModeration + ?Sized,
158 {
159 crate::moderate::timeout_user(moderation, actor, room, target, duration).await
160 }
161
162 /// Drop rate-limit buckets nobody has touched recently.
163 ///
164 /// The host must call this on a timer. Bucket state is keyed by (user, room),
165 /// which is unbounded and attacker-influenced, and both consumers run under a
166 /// hard 512M cgroup cap. Uses the shortest interval that cannot turn eviction
167 /// into a way to skip the queue.
168 ///
169 /// Returns how many buckets were dropped.
170 pub fn sweep(&self, now: Instant) -> usize {
171 self.limiter
172 .evict_idle(now, self.limiter.recommended_idle_timeout())
173 }
174
175 /// How often [`Chat::sweep`] wants to be called.
176 pub fn sweep_interval(&self) -> Duration {
177 self.limiter.recommended_idle_timeout()
178 }
179 }
180
181 #[cfg(test)]
182 mod tests {
183 use super::*;
184 use crate::retention::Retention;
185 use crate::room::RoomState;
186 use crate::traits::WriteAccess;
187 use async_trait::async_trait;
188 use uuid::Uuid;
189
190 struct Yes;
191
192 #[async_trait]
193 impl ChatAuthz for Yes {
194 async fn can_read(&self, _v: Option<UserId>, _r: &Room) -> Result<bool, ChatError> {
195 Ok(true)
196 }
197 async fn can_write(&self, _u: UserId, _r: &Room) -> Result<WriteAccess, ChatError> {
198 Ok(WriteAccess::Allow)
199 }
200 async fn is_moderator(&self, _u: UserId, _r: &Room) -> Result<bool, ChatError> {
201 Ok(false)
202 }
203 }
204
205 fn room() -> Room {
206 Room {
207 id: RoomId(Uuid::new_v4()),
208 state: RoomState::Open,
209 retention: Retention::forum_default(),
210 }
211 }
212
213 #[tokio::test]
214 async fn send_and_subscribe_share_the_same_rooms() {
215 let chat = Chat::new(HubLimits::default(), RateLimits::default());
216 let r = room();
217 let author = UserId(Uuid::new_v4());
218
219 let mut stream = chat
220 .subscribe(r.id, UserId(Uuid::new_v4()), None, |_| async { Ok(vec![]) })
221 .await
222 .unwrap();
223
224 chat.send(
225 &Yes,
226 SendRequest {
227 room: &r,
228 author,
229 body: "hi",
230 nonce: None,
231 now: Instant::now(),
232 },
233 |body| async move {
234 Ok(Message {
235 id: MessageId(1),
236 room_id: r.id,
237 author_id: author,
238 body_html: body.as_str().to_owned(),
239 created_at: 0,
240 nonce: None,
241 author: None,
242 })
243 },
244 )
245 .await
246 .unwrap();
247
248 assert!(
249 matches!(
250 stream.next().await,
251 Some(crate::event::ChatEvent::Message(_))
252 ),
253 "a message sent through Chat reaches a stream opened through Chat"
254 );
255 }
256
257 #[test]
258 fn sweep_uses_an_interval_that_is_not_a_bypass() {
259 let chat = Chat::new(
260 HubLimits::default(),
261 RateLimits {
262 burst: 5,
263 sustain_per_min: 60,
264 },
265 );
266 assert_eq!(chat.sweep_interval(), Duration::from_secs(5));
267
268 let t0 = Instant::now();
269 chat.limiter()
270 .check(UserId(Uuid::nil()), RoomId(Uuid::nil()), t0);
271 assert_eq!(chat.sweep(t0), 0, "a fresh bucket is not idle");
272 assert_eq!(chat.sweep(t0 + Duration::from_secs(10)), 1);
273 }
274 }
275