Skip to main content

max / makenotwork

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