1use std::collections::{BTreeMap, BTreeSet, VecDeque};
2use std::sync::{Arc, Mutex, MutexGuard, PoisonError};
3
4use crate::board::split_run;
5use crate::hash::draw;
6use crate::{
7 Board, BoardInfo, Change, Delivery, Display, Entry, Error, Event, Fake, Lobby, LobbyFilter,
8 LobbyKind, Message, Range, Sort, Ugc, Upload, Uploaded, User,
9};
10
11#[derive(Clone, Debug, Default)]
12pub struct Hub {
13 world: Arc<Mutex<World>>,
14}
15
16#[derive(Debug, Default)]
17pub(crate) struct World {
18 online: BTreeSet<User>,
19 events: BTreeMap<User, VecDeque<Event>>,
20 server_time: Option<u64>,
21 friends: BTreeSet<(User, User)>,
22 boards: Vec<BoardState>,
23 lobbies: BTreeMap<Lobby, LobbyState>,
24 lobby_count: u64,
25 sessions: BTreeMap<(User, User), Session>,
26 mailbox: BTreeMap<(User, u32), VecDeque<Message>>,
27 loss: Option<Loss>,
28 tick: u64,
29 ugc: BTreeMap<Ugc, Vec<u8>>,
30 ugc_count: u64,
31}
32
33#[derive(Debug)]
34struct BoardState {
35 name: String,
36 sort: Sort,
37 display: Display,
38 scores: BTreeMap<User, Scored>,
39}
40
41#[derive(Debug)]
42struct Scored {
43 score: i32,
44 details: Vec<i32>,
45 tick: u64,
46 attached: Option<Ugc>,
47}
48
49#[derive(Debug)]
50struct LobbyState {
51 kind: LobbyKind,
52 max: u32,
53 owner: User,
54 members: Vec<User>,
55 data: BTreeMap<String, String>,
56}
57
58#[derive(Debug)]
59enum Session {
60 Requested {
61 from: User,
62 queued: Vec<(u32, Vec<u8>)>,
63 },
64 Open,
65}
66
67#[derive(Debug)]
68struct Loss {
69 state: u64,
70 per_mille: u32,
71}
72
73fn pair(a: User, b: User) -> (User, User) {
74 if a < b { (a, b) } else { (b, a) }
75}
76
77impl Hub {
78 pub fn new() -> Self {
79 Self::default()
80 }
81
82 pub fn client(&self, user: User) -> Fake {
83 Fake::on(self, user)
84 }
85
86 pub fn set_server_time(&self, seconds: Option<u64>) {
87 self.world().server_time = seconds;
88 }
89
90 pub fn befriend(&self, a: User, b: User) {
91 if a != b {
92 self.world().friends.insert(pair(a, b));
93 }
94 }
95
96 pub fn set_loss(&self, seed: u64, per_mille: u32) {
97 self.world().loss = (per_mille > 0).then_some(Loss {
98 state: seed,
99 per_mille: per_mille.min(1000),
100 });
101 }
102
103 pub fn invite(&self, from: User, to: User, lobby: Lobby) {
104 self.world().push(
105 to,
106 Event::JoinRequested {
107 lobby,
108 friend: Some(from),
109 },
110 );
111 }
112
113 pub fn fail_session(&self, a: User, b: User, reason: Option<i32>) {
114 let mut world = self.world();
115 world.sessions.remove(&pair(a, b));
116 world.push(a, Event::SessionFailed { peer: b, reason });
117 world.push(b, Event::SessionFailed { peer: a, reason });
118 }
119
120 pub fn attached(&self, board: Board, user: User) -> Option<Ugc> {
121 let world = self.world();
122 let state = world.board(board).ok()?;
123 state.scores.get(&user).and_then(|scored| scored.attached)
124 }
125
126 pub(crate) fn world(&self) -> MutexGuard<'_, World> {
127 self.world.lock().unwrap_or_else(PoisonError::into_inner)
128 }
129}
130
131impl World {
132 pub(crate) fn arrive(&mut self, user: User) {
133 self.online.insert(user);
134 }
135
136 pub(crate) fn push(&mut self, user: User, event: Event) {
137 if self.online.contains(&user) {
138 self.events.entry(user).or_default().push_back(event);
139 }
140 }
141
142 pub(crate) fn take(&mut self, user: User) -> Vec<Event> {
143 self.events
144 .get_mut(&user)
145 .map(|queue| queue.drain(..).collect())
146 .unwrap_or_default()
147 }
148
149 pub(crate) fn server_time(&self) -> Option<u64> {
150 self.server_time
151 }
152
153 pub(crate) fn depart(&mut self, user: User) {
154 let joined: Vec<Lobby> = self
155 .lobbies
156 .iter()
157 .filter(|(_, lobby)| lobby.members.contains(&user))
158 .map(|(id, _)| *id)
159 .collect();
160 for lobby in joined {
161 self.leave(user, lobby, Change::Disconnected);
162 }
163 let peers: Vec<(User, User)> = self
164 .sessions
165 .keys()
166 .filter(|(a, b)| *a == user || *b == user)
167 .copied()
168 .collect();
169 for key in peers {
170 self.sessions.remove(&key);
171 let peer = if key.0 == user { key.1 } else { key.0 };
172 self.push(
173 peer,
174 Event::SessionFailed {
175 peer: user,
176 reason: None,
177 },
178 );
179 }
180 self.online.remove(&user);
181 self.events.remove(&user);
182 self.mailbox.retain(|(owner, _), _| *owner != user);
183 }
184
185 fn info(&self, index: usize) -> BoardInfo {
186 let board = &self.boards[index];
187 BoardInfo {
188 board: Board(index as u64 + 1),
189 name: board.name.clone(),
190 sort: board.sort,
191 display: board.display,
192 entries: board.scores.len() as u32,
193 }
194 }
195
196 pub(crate) fn find_board(&self, name: &str) -> Result<BoardInfo, Error> {
197 self.boards
198 .iter()
199 .position(|board| board.name == name)
200 .map(|index| self.info(index))
201 .ok_or(Error::Unknown)
202 }
203
204 pub(crate) fn create_board(&mut self, name: &str, sort: Sort, display: Display) -> BoardInfo {
205 if let Ok(info) = self.find_board(name) {
206 return info;
207 }
208 self.boards.push(BoardState {
209 name: name.to_string(),
210 sort,
211 display,
212 scores: BTreeMap::new(),
213 });
214 self.info(self.boards.len() - 1)
215 }
216
217 fn board(&self, board: Board) -> Result<&BoardState, Error> {
218 usize::try_from(board.0)
219 .ok()
220 .and_then(|id| id.checked_sub(1))
221 .and_then(|index| self.boards.get(index))
222 .ok_or(Error::Unknown)
223 }
224
225 fn ranking(board: &BoardState) -> Vec<(User, &Scored)> {
226 let mut rows: Vec<(User, &Scored)> = board
227 .scores
228 .iter()
229 .map(|(user, scored)| (*user, scored))
230 .collect();
231 rows.sort_by(|(a_user, a), (b_user, b)| {
232 let by_score = match board.sort {
233 Sort::Ascending => a.score.cmp(&b.score),
234 Sort::Descending => b.score.cmp(&a.score),
235 };
236 by_score.then(a.tick.cmp(&b.tick)).then(a_user.cmp(b_user))
237 });
238 rows
239 }
240
241 fn rank(board: &BoardState, user: User) -> u32 {
242 Self::ranking(board)
243 .iter()
244 .position(|(who, _)| *who == user)
245 .map(|index| index as u32 + 1)
246 .unwrap_or(0)
247 }
248
249 pub(crate) fn upload(
250 &mut self,
251 user: User,
252 board: Board,
253 score: i32,
254 details: &[i32],
255 upload: Upload,
256 ) -> Result<Uploaded, Error> {
257 self.board(board)?;
258 self.tick += 1;
259 let tick = self.tick;
260 let index = board.0 as usize - 1;
261 let state = &mut self.boards[index];
262 let previous_rank = Self::rank(state, user);
263 let changed = match (upload, state.scores.get(&user)) {
264 (Upload::Force, _) | (Upload::KeepBest, None) => true,
265 (Upload::KeepBest, Some(old)) => state.sort.better(score, old.score),
266 };
267 if changed {
268 state.scores.insert(
269 user,
270 Scored {
271 score,
272 details: details.to_vec(),
273 tick,
274 attached: None,
275 },
276 );
277 }
278 Ok(Uploaded {
279 score,
280 changed,
281 rank: Self::rank(state, user),
282 previous_rank,
283 })
284 }
285
286 pub(crate) fn download(
287 &self,
288 user: User,
289 board: Board,
290 range: Range,
291 details: usize,
292 ) -> Result<Vec<Entry>, Error> {
293 let state = self.board(board)?;
294 let rows = Self::ranking(state);
295 let picked: Vec<usize> = match range {
296 Range::Global { first, last } => {
297 let first = first as usize - 1;
298 let last = (last as usize).min(rows.len());
299 (first..last).collect()
300 }
301 Range::AroundUser { before, after } => {
302 match rows.iter().position(|(who, _)| *who == user) {
303 Some(at) => {
304 let first = at.saturating_sub(before as usize);
305 let last = at.saturating_add(after as usize).min(rows.len() - 1);
306 (first..=last).collect()
307 }
308 None => Vec::new(),
309 }
310 }
311 Range::Friends => (0..rows.len())
312 .filter(|index| {
313 let who = rows[*index].0;
314 who == user || self.friends.contains(&pair(who, user))
315 })
316 .collect(),
317 };
318 Ok(picked
319 .into_iter()
320 .map(|index| {
321 let (who, scored) = rows[index];
322 let (run, details) = split_run(&scored.details, details);
323 Entry {
324 user: who,
325 rank: index as u32 + 1,
326 score: scored.score,
327 details,
328 run,
329 }
330 })
331 .collect())
332 }
333
334 pub(crate) fn share(&mut self, bytes: Vec<u8>) -> Ugc {
335 self.ugc_count += 1;
336 let ugc = Ugc(0x0100_0000_0000_0000 + self.ugc_count);
337 self.ugc.insert(ugc, bytes);
338 ugc
339 }
340
341 pub(crate) fn ugc(&self, ugc: Ugc) -> Result<Vec<u8>, Error> {
342 self.ugc.get(&ugc).cloned().ok_or(Error::Missing)
343 }
344
345 pub(crate) fn attach(&mut self, user: User, board: Board, ugc: Ugc) -> Result<(), Error> {
346 self.board(board)?;
347 if !self.ugc.contains_key(&ugc) {
348 return Err(Error::Missing);
349 }
350 let state = &mut self.boards[board.0 as usize - 1];
351 let scored = state.scores.get_mut(&user).ok_or(Error::Rejected)?;
352 scored.attached = Some(ugc);
353 Ok(())
354 }
355
356 pub(crate) fn create_lobby(&mut self, user: User, kind: LobbyKind, max: u32) -> Lobby {
357 self.lobby_count += 1;
358 let lobby = Lobby(self.lobby_count);
359 self.lobbies.insert(
360 lobby,
361 LobbyState {
362 kind,
363 max,
364 owner: user,
365 members: vec![user],
366 data: BTreeMap::new(),
367 },
368 );
369 lobby
370 }
371
372 pub(crate) fn find_lobbies(&self, filter: &LobbyFilter) -> Vec<Lobby> {
373 let mut found: Vec<(Vec<u64>, Lobby)> = self
374 .lobbies
375 .iter()
376 .filter(|(_, state)| matches!(state.kind, LobbyKind::Public | LobbyKind::Invisible))
377 .filter(|(_, state)| {
378 let open = state.max.saturating_sub(state.members.len() as u32);
379 filter.matches(&state.data, open)
380 })
381 .map(|(lobby, state)| (filter.nearness(&state.data), *lobby))
382 .collect();
383 found.sort();
384 found
385 .into_iter()
386 .map(|(_, lobby)| lobby)
387 .take(filter.limit())
388 .collect()
389 }
390
391 pub(crate) fn join(&mut self, user: User, lobby: Lobby) -> Result<Lobby, Error> {
392 let state = self.lobbies.get_mut(&lobby).ok_or(Error::Rejected)?;
393 if state.members.contains(&user) {
394 return Ok(lobby);
395 }
396 if state.members.len() as u32 >= state.max {
397 return Err(Error::Rejected);
398 }
399 let others = state.members.clone();
400 state.members.push(user);
401 for member in others {
402 self.push(
403 member,
404 Event::Member {
405 lobby,
406 user,
407 change: Change::Entered,
408 },
409 );
410 }
411 Ok(lobby)
412 }
413
414 pub(crate) fn leave(&mut self, user: User, lobby: Lobby, change: Change) {
415 let Some(state) = self.lobbies.get_mut(&lobby) else {
416 return;
417 };
418 let Some(at) = state.members.iter().position(|member| *member == user) else {
419 return;
420 };
421 state.members.remove(at);
422 if state.members.is_empty() {
423 self.lobbies.remove(&lobby);
424 return;
425 }
426 if state.owner == user {
427 state.owner = state.members[0];
428 }
429 let others = state.members.clone();
430 for member in others {
431 self.push(
432 member,
433 Event::Member {
434 lobby,
435 user,
436 change,
437 },
438 );
439 }
440 }
441
442 fn member_of(&self, user: User, lobby: Lobby) -> Option<&LobbyState> {
443 self.lobbies
444 .get(&lobby)
445 .filter(|state| state.members.contains(&user))
446 }
447
448 pub(crate) fn members(&self, user: User, lobby: Lobby) -> Vec<User> {
449 self.member_of(user, lobby)
450 .map(|state| state.members.clone())
451 .unwrap_or_default()
452 }
453
454 pub(crate) fn owner(&self, user: User, lobby: Lobby) -> Result<User, Error> {
455 self.member_of(user, lobby)
456 .map(|state| state.owner)
457 .ok_or(Error::Rejected)
458 }
459
460 pub(crate) fn set_lobby_data(
461 &mut self,
462 user: User,
463 lobby: Lobby,
464 key: &str,
465 value: &str,
466 ) -> Result<(), Error> {
467 let state = self
468 .lobbies
469 .get_mut(&lobby)
470 .filter(|state| state.owner == user)
471 .ok_or(Error::Rejected)?;
472 if value.is_empty() {
473 state.data.remove(key);
474 } else {
475 state.data.insert(key.to_string(), value.to_string());
476 }
477 let members = state.members.clone();
478 for member in members {
479 self.push(member, Event::LobbyData { lobby });
480 }
481 Ok(())
482 }
483
484 pub(crate) fn lobby_data(&self, lobby: Lobby, key: &str) -> Option<String> {
485 self.lobbies
486 .get(&lobby)
487 .and_then(|state| state.data.get(key).cloned())
488 }
489
490 fn dropped(&mut self) -> bool {
491 let Some(loss) = self.loss.as_mut() else {
492 return false;
493 };
494 draw(&mut loss.state) % 1000 < u64::from(loss.per_mille)
495 }
496
497 fn deliver(&mut self, from: User, to: User, channel: u32, bytes: Vec<u8>) {
498 self.mailbox
499 .entry((to, channel))
500 .or_default()
501 .push_back(Message {
502 peer: from,
503 channel,
504 bytes,
505 });
506 }
507
508 fn open(&mut self, from: User, to: User, queued: Vec<(u32, Vec<u8>)>) {
509 self.sessions.insert(pair(from, to), Session::Open);
510 for (channel, bytes) in queued {
511 self.deliver(from, to, channel, bytes);
512 }
513 self.push(from, Event::Connected { peer: to });
514 self.push(to, Event::Connected { peer: from });
515 }
516
517 pub(crate) fn send(
518 &mut self,
519 user: User,
520 peer: User,
521 channel: u32,
522 bytes: &[u8],
523 delivery: Delivery,
524 ) {
525 if !self.online.contains(&peer) {
526 self.push(user, Event::SessionFailed { peer, reason: None });
527 return;
528 }
529 if delivery == Delivery::Unreliable && self.dropped() {
530 return;
531 }
532 let key = pair(user, peer);
533 match self.sessions.remove(&key) {
534 None => {
535 self.sessions.insert(
536 key,
537 Session::Requested {
538 from: user,
539 queued: vec![(channel, bytes.to_vec())],
540 },
541 );
542 self.push(peer, Event::SessionRequest { peer: user });
543 }
544 Some(Session::Requested { from, mut queued }) if from == user => {
545 queued.push((channel, bytes.to_vec()));
546 self.sessions
547 .insert(key, Session::Requested { from, queued });
548 }
549 Some(Session::Requested { queued, .. }) => {
550 self.open(peer, user, queued);
551 self.deliver(user, peer, channel, bytes.to_vec());
552 }
553 Some(Session::Open) => {
554 self.sessions.insert(key, Session::Open);
555 self.deliver(user, peer, channel, bytes.to_vec());
556 }
557 }
558 }
559
560 pub(crate) fn accept(&mut self, user: User, peer: User) -> Result<(), Error> {
561 let key = pair(user, peer);
562 match self.sessions.remove(&key) {
563 Some(Session::Requested { from, queued }) if from == peer => {
564 self.open(peer, user, queued);
565 Ok(())
566 }
567 Some(session) => {
568 self.sessions.insert(key, session);
569 Err(Error::Missing)
570 }
571 None => Err(Error::Missing),
572 }
573 }
574
575 pub(crate) fn receive(&mut self, user: User, channel: u32, max: usize) -> Vec<Message> {
576 let Some(queue) = self.mailbox.get_mut(&(user, channel)) else {
577 return Vec::new();
578 };
579 let count = max.min(queue.len());
580 queue.drain(..count).collect()
581 }
582}