manabrew_server/
draft_clock.rs1use std::sync::Arc;
2use std::time::{Duration, Instant};
3
4use serde::{Deserialize, Serialize};
5
6use crate::chat::unix_time_ms;
7use crate::connection::{broadcast_to_room, send_to_room_player};
8use crate::protocol::ServerMessage;
9use crate::room::Room;
10use crate::state::ServerState;
11
12#[derive(Debug, Clone, Deserialize, Serialize)]
13#[serde(rename_all = "camelCase", deny_unknown_fields)]
14pub struct ClockSeat {
15 pub seat: u8,
16 pub revision: String,
17 pub remaining_ms: u64,
18 pub deadline_ms: Option<u64>,
19 #[serde(skip)]
20 deadline: Option<Instant>,
21 #[serde(skip)]
22 expired: bool,
23}
24
25#[derive(Debug, Deserialize)]
26#[serde(rename_all = "camelCase", deny_unknown_fields)]
27struct ClockSync {
28 #[serde(rename = "type")]
29 kind: String,
30 session_id: String,
31 sequence: u64,
32 paused: bool,
33 seats: Vec<ClockSeat>,
34}
35
36#[derive(Debug)]
37pub struct DraftClock {
38 session_id: String,
39 sequence: u64,
40 paused: bool,
41 seats: Vec<ClockSeat>,
42}
43
44fn envelope(room: &Room, payload: serde_json::Value, target: Option<&str>) -> ServerMessage {
45 ServerMessage::StateUpdate {
46 from_player: room.host_username.clone(),
47 state: serde_json::json!({
48 "kind": "roomRelay", "protocol": "draft-v1", "version": 1,
49 "messageId": uuid::Uuid::new_v4().to_string(),
50 "roomId": room.room_id, "fromPlayer": room.host_username,
51 "targetPlayer": target, "payload": payload,
52 }),
53 }
54}
55
56fn state_message(room: &Room) -> Option<ServerMessage> {
57 let clock = room.draft_clock.as_ref()?;
58 Some(envelope(
59 room,
60 serde_json::json!({
61 "type": "clockState", "sessionId": clock.session_id,
62 "sequence": clock.sequence, "paused": clock.paused,
63 "serverNowMs": unix_time_ms(), "seats": clock.seats,
64 }),
65 None,
66 ))
67}
68
69pub fn synchronize(state: &Arc<ServerState>, room_id: &str, payload: &serde_json::Value) {
70 let Ok(sync) = serde_json::from_value::<ClockSync>(payload.clone()) else {
71 return;
72 };
73 if sync.kind != "clockSync" || sync.seats.len() > 8 {
74 return;
75 }
76 let (message, timers) = {
77 let Some(mut room) = state.rooms.get_mut(room_id) else {
78 return;
79 };
80 if room.limited_session_id.as_deref() != Some(sync.session_id.as_str())
81 || !room.host_connected()
82 || room
83 .draft_config
84 .as_ref()
85 .and_then(|config| config.pick_seconds)
86 .is_none()
87 {
88 return;
89 }
90 if let Some(clock) = &room.draft_clock {
91 if sync.sequence <= clock.sequence {
92 let message = state_message(&room);
93 drop(room);
94 if let Some(message) = message {
95 broadcast_to_room(state, room_id, &message);
96 }
97 return;
98 }
99 }
100 let duration = u64::from(
101 room.draft_config
102 .as_ref()
103 .and_then(|config| config.pick_seconds)
104 .unwrap(),
105 ) * 1000;
106 let mut seen = [false; 8];
107 if sync.seats.iter().any(|seat| {
108 seat.seat >= room.max_players
109 || seat.revision.len() > 64
110 || seat.revision.is_empty()
111 || seat.remaining_ms > duration
112 || std::mem::replace(&mut seen[usize::from(seat.seat)], true)
113 }) {
114 return;
115 }
116 let now = Instant::now();
117 let wall_now = unix_time_ms();
118 let mut seats = sync.seats;
119 let mut timers = Vec::with_capacity(seats.len());
120 for seat in &mut seats {
121 if let Some(previous) = room.draft_clock.as_ref().and_then(|clock| {
122 clock
123 .seats
124 .iter()
125 .find(|entry| entry.seat == seat.seat && entry.revision == seat.revision)
126 }) {
127 let retained = previous
128 .deadline
129 .map(|deadline| deadline.saturating_duration_since(now).as_millis() as u64)
130 .unwrap_or(previous.remaining_ms);
131 seat.remaining_ms = seat.remaining_ms.min(retained);
132 }
133 seat.expired = false;
134 seat.deadline = (!sync.paused).then(|| now + Duration::from_millis(seat.remaining_ms));
135 seat.deadline_ms = (!sync.paused).then(|| wall_now + seat.remaining_ms);
136 if !sync.paused {
137 timers.push((seat.seat, seat.revision.clone(), seat.remaining_ms));
138 }
139 }
140 room.draft_clock = Some(DraftClock {
141 session_id: sync.session_id.clone(),
142 sequence: sync.sequence,
143 paused: sync.paused,
144 seats,
145 });
146 (state_message(&room), timers)
147 };
148 if let Some(message) = message {
149 broadcast_to_room(state, room_id, &message);
150 }
151 for (seat, revision, delay) in timers {
152 let state = state.clone();
153 let room_id = room_id.to_string();
154 let session_id = sync.session_id.clone();
155 tokio::spawn(async move {
156 tokio::time::sleep(Duration::from_millis(delay)).await;
157 let message = {
158 let Some(mut room) = state.rooms.get_mut(&room_id) else {
159 return;
160 };
161 if !room.host_connected() {
162 return;
163 }
164 let Some(clock) = room.draft_clock.as_mut() else {
165 return;
166 };
167 if clock.session_id != session_id || clock.sequence != sync.sequence || clock.paused
168 {
169 return;
170 }
171 let Some(current) = clock
172 .seats
173 .iter_mut()
174 .find(|entry| entry.seat == seat && entry.revision == revision)
175 else {
176 return;
177 };
178 if current.expired
179 || current
180 .deadline
181 .is_none_or(|deadline| deadline > Instant::now())
182 {
183 return;
184 }
185 current.expired = true;
186 current.remaining_ms = 0;
187 current.deadline = None;
188 current.deadline_ms = None;
189 envelope(
190 &room,
191 serde_json::json!({ "type": "clockExpired", "sessionId": session_id, "sequence": sync.sequence, "seat": seat, "revision": revision }),
192 Some(&room.host_username),
193 )
194 };
195 let host = state
196 .rooms
197 .get(&room_id)
198 .map(|room| room.host_username.clone());
199 if let Some(host) = host {
200 send_to_room_player(&state, &room_id, &host, &message);
201 }
202 });
203 }
204}
205
206pub fn pause_disconnected_host(state: &Arc<ServerState>, room_id: &str, player_id: &str) {
207 let message = {
208 let Some(mut room) = state.rooms.get_mut(room_id) else {
209 return;
210 };
211 if !room.is_host(player_id) {
212 return;
213 }
214 let Some(clock) = room.draft_clock.as_mut() else {
215 return;
216 };
217 if clock.paused {
218 return;
219 }
220 clock.paused = true;
221 let now = Instant::now();
222 for seat in &mut clock.seats {
223 seat.remaining_ms = seat
224 .deadline
225 .map(|deadline| deadline.saturating_duration_since(now).as_millis() as u64)
226 .unwrap_or(seat.remaining_ms);
227 seat.deadline = None;
228 seat.deadline_ms = None;
229 }
230 state_message(&room)
231 };
232 if let Some(message) = message {
233 broadcast_to_room(state, room_id, &message);
234 }
235}
236
237pub fn replay(state: &Arc<ServerState>, room_id: &str, username: &str) {
238 let message = state
239 .rooms
240 .get(room_id)
241 .and_then(|room| state_message(&room));
242 if let Some(message) = message {
243 send_to_room_player(state, room_id, username, &message);
244 }
245}