netget/server/matrix/
hub.rs

1//! The homeserver's protocol state: access tokens, rooms and their members, and per-user
2//! delivery queues for `/sync`. This is transport state — what a homeserver must hold to
3//! hand one client's events to another — not a store: what a room *says* comes from the
4//! clients and the model, and nothing outlives the server.
5use serde_json::{json, Map, Value};
6use std::collections::{BTreeSet, HashMap, VecDeque};
7
8/// Items waiting in one user's queue; past this the oldest go and the next sync is `limited`.
9pub const MAX_QUEUE: usize = 1000;
10/// Events a room keeps (what a joining user and `/messages` are given).
11pub const ROOM_HISTORY: usize = 200;
12pub const MAX_ROOMS: usize = 1000;
13pub const MAX_TOKENS: usize = 10_000;
14
15pub struct Room {
16    pub name: Option<String>,
17    pub members: BTreeSet<String>,
18    pub invited: BTreeSet<String>,
19    pub history: VecDeque<Value>,
20}
21
22#[derive(Clone)]
23enum Item {
24    Timeline(Value),
25    /// The stripped state an invitee is shown.
26    Invite(Vec<Value>),
27}
28
29#[derive(Default)]
30pub struct Hub {
31    pub server_name: String,
32    tokens: HashMap<String, (String, String)>,
33    pub rooms: HashMap<String, Room>,
34    queues: HashMap<String, VecDeque<(u64, String, Item)>>,
35    dropped: HashMap<String, bool>,
36    seq: u64,
37}
38
39pub fn random_id(n: usize) -> String {
40    use rand::Rng;
41    const A: &[u8] = b"ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz";
42    let mut r = rand::thread_rng();
43    (0..n).map(|_| A[r.gen_range(0..A.len())] as char).collect()
44}
45
46pub fn now_ms() -> u64 {
47    crate::utils::clock::SystemTime::now()
48        .duration_since(crate::utils::clock::UNIX_EPOCH)
49        .map(|d| d.as_millis() as u64)
50        .unwrap_or(0)
51}
52
53/// `s12` → 12; anything else → None (a fresh, full sync).
54pub fn parse_batch(since: Option<&str>) -> Option<u64> {
55    since
56        .and_then(|s| s.strip_prefix('s'))
57        .and_then(|n| n.parse().ok())
58}
59
60impl Hub {
61    pub fn new(server_name: &str) -> Self {
62        Self {
63            server_name: server_name.to_string(),
64            ..Default::default()
65        }
66    }
67
68    /// `alice` or `@alice:server` → `@alice:server`.
69    pub fn user_id(&self, user: &str) -> String {
70        if user.starts_with('@') {
71            user.to_string()
72        } else {
73            format!("@{}:{}", user.to_lowercase(), self.server_name)
74        }
75    }
76
77    pub fn issue_token(&mut self, user_id: &str, device_id: &str) -> Option<String> {
78        if self.tokens.len() >= MAX_TOKENS {
79            return None;
80        }
81        let token = format!("ngt_{}", random_id(32));
82        self.tokens
83            .insert(token.clone(), (user_id.to_string(), device_id.to_string()));
84        Some(token)
85    }
86
87    pub fn whoami(&self, token: &str) -> Option<(String, String)> {
88        self.tokens.get(token).cloned()
89    }
90
91    pub fn logout(&mut self, token: &str) {
92        self.tokens.remove(token);
93    }
94
95    pub fn room_name(&self, room_id: &str) -> Option<String> {
96        self.rooms.get(room_id).and_then(|r| r.name.clone())
97    }
98
99    pub fn is_member(&self, room_id: &str, user: &str) -> bool {
100        self.rooms
101            .get(room_id)
102            .is_some_and(|r| r.members.contains(user))
103    }
104
105    pub fn members(&self, room_id: &str) -> Vec<String> {
106        self.rooms
107            .get(room_id)
108            .map(|r| r.members.iter().cloned().collect())
109            .unwrap_or_default()
110    }
111
112    fn enqueue(&mut self, user: &str, room_id: &str, item: Item) {
113        self.seq += 1;
114        let seq = self.seq;
115        let q = self.queues.entry(user.to_string()).or_default();
116        q.push_back((seq, room_id.to_string(), item));
117        while q.len() > MAX_QUEUE {
118            q.pop_front();
119            self.dropped.insert(user.to_string(), true);
120        }
121    }
122
123    /// A room with its creator joined and the `m.room.create`, membership and name events a
124    /// client needs to show it; invitees are invited.
125    pub fn create_room(
126        &mut self,
127        creator: &str,
128        name: Option<String>,
129        invite: &[String],
130    ) -> Option<String> {
131        if self.rooms.len() >= MAX_ROOMS {
132            return None;
133        }
134        let id = format!("!{}:{}", random_id(18), self.server_name);
135        self.rooms.insert(
136            id.clone(),
137            Room {
138                name: name.clone(),
139                members: BTreeSet::from([creator.to_string()]),
140                invited: BTreeSet::new(),
141                history: VecDeque::new(),
142            },
143        );
144        self.post(
145            &id,
146            creator,
147            "m.room.create",
148            json!({"creator": creator, "room_version": "10"}),
149            Some(""),
150        );
151        self.post(
152            &id,
153            creator,
154            "m.room.member",
155            json!({"membership": "join"}),
156            Some(creator),
157        );
158        self.post(
159            &id,
160            creator,
161            "m.room.join_rules",
162            json!({"join_rule": "invite"}),
163            Some(""),
164        );
165        if let Some(n) = name {
166            self.post(&id, creator, "m.room.name", json!({"name": n}), Some(""));
167        }
168        for user in invite {
169            self.invite(&id, creator, user);
170        }
171        Some(id)
172    }
173
174    pub fn invite(&mut self, room_id: &str, sender: &str, user: &str) -> bool {
175        let Some(room) = self.rooms.get_mut(room_id) else {
176            return false;
177        };
178        if room.members.contains(user) {
179            return true;
180        }
181        room.invited.insert(user.to_string());
182        let name = room.name.clone();
183        self.post(
184            room_id,
185            sender,
186            "m.room.member",
187            json!({"membership": "invite"}),
188            Some(user),
189        );
190        let mut stripped = vec![
191            json!({"type": "m.room.member", "state_key": user, "sender": sender, "content": {"membership": "invite"}}),
192            json!({"type": "m.room.join_rules", "state_key": "", "sender": sender, "content": {"join_rule": "invite"}}),
193        ];
194        if let Some(n) = name {
195            stripped.push(json!({"type": "m.room.name", "state_key": "", "sender": sender, "content": {"name": n}}));
196        }
197        self.enqueue(user, room_id, Item::Invite(stripped));
198        true
199    }
200
201    /// Add an event to a room and every member's queue; returns its id.
202    pub fn post(
203        &mut self,
204        room_id: &str,
205        sender: &str,
206        kind: &str,
207        content: Value,
208        state_key: Option<&str>,
209    ) -> Option<String> {
210        let event_id = format!("${}", random_id(32));
211        let mut event = json!({
212            "type": kind,
213            "content": content,
214            "sender": sender,
215            "event_id": event_id,
216            "room_id": room_id,
217            "origin_server_ts": now_ms(),
218            "unsigned": {},
219        });
220        if let Some(k) = state_key {
221            event["state_key"] = json!(k);
222        }
223        let members: Vec<String> = {
224            let room = self.rooms.get_mut(room_id)?;
225            room.history.push_back(event.clone());
226            while room.history.len() > ROOM_HISTORY {
227                room.history.pop_front();
228            }
229            room.members.iter().cloned().collect()
230        };
231        for m in members {
232            self.enqueue(&m, room_id, Item::Timeline(event.clone()));
233        }
234        Some(event_id)
235    }
236
237    /// Join `user` to a room: they are handed its history, then everyone sees them join.
238    pub fn join(&mut self, room_id: &str, user: &str) -> bool {
239        let history: Vec<Value> = match self.rooms.get_mut(room_id) {
240            Some(r) if r.members.contains(user) => return true,
241            Some(r) => {
242                r.invited.remove(user);
243                r.members.insert(user.to_string());
244                r.history.iter().cloned().collect()
245            }
246            None => return false,
247        };
248        for event in history {
249            self.enqueue(user, room_id, Item::Timeline(event));
250        }
251        self.post(
252            room_id,
253            user,
254            "m.room.member",
255            json!({"membership": "join"}),
256            Some(user),
257        );
258        true
259    }
260
261    /// Leave a room or decline an invite; everyone (the leaver included) sees it.
262    pub fn leave(&mut self, room_id: &str, user: &str) -> bool {
263        let (member, invited) = match self.rooms.get(room_id) {
264            Some(r) => (r.members.contains(user), r.invited.contains(user)),
265            None => return false,
266        };
267        if !member && !invited {
268            return false;
269        }
270        self.post(
271            room_id,
272            user,
273            "m.room.member",
274            json!({"membership": "leave"}),
275            Some(user),
276        );
277        if let Some(r) = self.rooms.get_mut(room_id) {
278            r.members.remove(user);
279            r.invited.remove(user);
280        }
281        true
282    }
283
284    pub fn joined(&self, user: &str) -> Vec<String> {
285        let mut v: Vec<String> = self
286            .rooms
287            .iter()
288            .filter(|(_, r)| r.members.contains(user))
289            .map(|(id, _)| id.clone())
290            .collect();
291        v.sort();
292        v
293    }
294
295    /// The `/sync` body for `user`, or None when nothing is new since batch `since`.
296    /// Items stay queued until the client acknowledges them by asking for a later batch, so a
297    /// response lost on the way is delivered again.
298    pub fn sync(&mut self, user: &str, since: Option<u64>) -> Option<Value> {
299        let seq = self.seq;
300        let q = self.queues.entry(user.to_string()).or_default();
301        if let Some(s) = since {
302            q.retain(|(n, _, _)| *n > s);
303        }
304        let new: Vec<(u64, String, Item)> = q.iter().cloned().collect();
305        let limited = self.dropped.remove(user).unwrap_or(false);
306        if new.is_empty() && since.is_some() && !limited {
307            return None;
308        }
309        let next = new.last().map(|(n, _, _)| *n).unwrap_or(seq);
310        let mut join = Map::new();
311        let mut invite = Map::new();
312        for (_, room, item) in new {
313            match item {
314                Item::Timeline(event) => {
315                    let entry = join.entry(room).or_insert_with(|| {
316                        json!({"timeline": {"events": [], "limited": limited, "prev_batch": format!("s{}", since.unwrap_or(0))},
317                               "state": {"events": []}, "ephemeral": {"events": []},
318                               "account_data": {"events": []}, "unread_notifications": {}})
319                    });
320                    if let Some(list) = entry["timeline"]["events"].as_array_mut() {
321                        list.push(event);
322                    }
323                }
324                Item::Invite(stripped) => {
325                    invite.insert(room, json!({"invite_state": {"events": stripped}}));
326                }
327            }
328        }
329        // A room the user has since joined is not still an invite.
330        invite.retain(|room, _| !self.is_member(room, user));
331        Some(json!({
332            "next_batch": format!("s{next}"),
333            "rooms": {"join": join, "invite": invite, "leave": {}},
334            "presence": {"events": []},
335            "account_data": {"events": []},
336            "to_device": {"events": []},
337            "device_lists": {"changed": [], "left": []},
338            "device_one_time_keys_count": {},
339        }))
340    }
341}
netget/server/matrix/hub.rs

netget/server/matrix/
hub.rs
