combat_service/
routes.rs

1use std::sync::Arc;
2
3use axum::{
4    Json, Router,
5    extract::{
6        Path, Query, State,
7        ws::{Message, WebSocket, WebSocketUpgrade},
8    },
9    http::{HeaderMap, StatusCode, header},
10    response::{IntoResponse, Response},
11    routing::{get, post},
12};
13use edda_auth::{
14    protocol::{Claims, Principal},
15    server::Authenticated,
16};
17use edda_types::combat::{
18    ClientMessage, CombatSnapshot, CombatState, CombatStatus, CombatSummary, CommandEnvelope,
19    Concealer, LoggedEvent, ServerMessage,
20};
21use futures_util::{SinkExt, StreamExt};
22use serde::Deserialize;
23use tokio::sync::broadcast::error::RecvError;
24use uuid::Uuid;
25
26use edda_server::openapi::ErrorBody;
27
28use crate::{
29    error::ApiError,
30    hub::{Hub, LiveCombat, Outcome},
31    store::ListFilter,
32};
33
34type AppState = Arc<Hub>;
35
36/// The `/combats` routes.
37pub fn router() -> Router<AppState> {
38    Router::new()
39        .route("/health", get(edda_server::health))
40        .route("/combats", get(list).post(create))
41        .route("/combats/{id}", get(snapshot))
42        .route("/combats/{id}/log", get(log))
43        .route("/combats/{id}/commands", post(command))
44        .route("/combats/{id}/ws", get(websocket))
45}
46
47/// `POST /combats`: a name, a plan, or both.
48#[derive(Deserialize, utoipa::ToSchema)]
49pub(crate) struct NewCombat {
50    /// What to call it; with a plan, the plan's name by default.
51    #[serde(default)]
52    name: Option<String>,
53    /// A planned encounter to start it from, with the plan's monsters in it.
54    #[serde(default)]
55    encounter: Option<Uuid>,
56}
57
58#[utoipa::path(
59    post,
60    path = "/combats",
61    tag = "combats",
62    request_body = NewCombat,
63    responses(
64        (status = 201, description = "Started.", body = CombatSnapshot, headers(("Location" = String, description = "Where it is: `/combats/{id}`."))),
65        (status = 400, description = "Neither a name nor a plan.", body = ErrorBody),
66        (status = 403, description = "Not a DM device, or the campaign has as many combats as it may keep.", body = ErrorBody),
67        (status = 404, description = "No such plan in the campaign.", body = ErrorBody),
68        (status = 422, description = "A monster in the plan is gone from the campaign.", body = ErrorBody),
69        (status = 502, description = "The monster service can't be reached.", body = ErrorBody),
70    )
71)]
72/// Creates a combat in the caller's campaign (DM devices only), empty or
73/// from a plan.
74async fn create(
75    State(hub): State<AppState>,
76    Authenticated(claims): Authenticated,
77    Json(body): Json<NewCombat>,
78) -> Result<impl IntoResponse, ApiError> {
79    if !claims.is_dm() {
80        return Err(ApiError::Forbidden(
81            "only a DM device can start a combat".into(),
82        ));
83    }
84    let snapshot = match (body.encounter, body.name) {
85        (Some(encounter), name) => {
86            hub.create_from_plan(
87                claims.campaign,
88                encounter,
89                name.as_deref(),
90                &claims.settings.quotas,
91            )
92            .await?
93        }
94        (None, Some(name)) if !name.trim().is_empty() => {
95            hub.create(&claims.campaign.to_string(), &name, &claims.settings.quotas)
96                .await?
97        }
98        (None, _) => return Err(ApiError::BadRequest("name must not be empty".into())),
99    };
100    let location = format!("/combats/{}", snapshot.id);
101    Ok((
102        StatusCode::CREATED,
103        [(header::LOCATION, location)],
104        Json(snapshot),
105    ))
106}
107
108#[derive(Deserialize, utoipa::IntoParams)]
109pub(crate) struct ListQuery {
110    /// Only combats with this status.
111    status: Option<CombatStatus>,
112}
113
114#[utoipa::path(
115    get,
116    path = "/combats",
117    tag = "combats",
118    params(ListQuery),
119    responses((status = 200, description = "The campaign's combats.", body = Vec<CombatSummary>))
120)]
121/// The caller's campaign's combats, most recently active first; filter with
122/// `?status=paused`.
123async fn list(
124    State(hub): State<AppState>,
125    Authenticated(claims): Authenticated,
126    Query(query): Query<ListQuery>,
127) -> Result<Json<Vec<CombatSummary>>, ApiError> {
128    let filter = ListFilter {
129        campaign: Some(claims.campaign.to_string()),
130        status: query.status,
131    };
132    Ok(Json(hub.store().list(&filter).await?))
133}
134
135#[utoipa::path(
136    get,
137    path = "/combats/{id}",
138    tag = "combats",
139    params(("id" = Uuid, Path, description = "The combat's id.")),
140    responses(
141        (status = 200, description = "Where the combat stands, with `seq`, the last event it includes. A player's device sees monsters' numbers concealed.", body = CombatSnapshot),
142        (status = 404, description = "No such combat in the campaign.", body = ErrorBody),
143    )
144)]
145/// Current state: round, whose turn it is, initiative order and effects.
146async fn snapshot(
147    State(hub): State<AppState>,
148    Authenticated(claims): Authenticated,
149    Path(id): Path<Uuid>,
150) -> Result<Json<CombatSnapshot>, ApiError> {
151    let combat = hub.combat_in(id, claims.campaign).await?;
152    let mut snapshot = combat.snapshot(hub.store()).await?;
153    if behind_the_screen(&claims) {
154        snapshot.state = Concealer::new(&snapshot.state).state(&snapshot.state);
155    }
156    Ok(Json(snapshot))
157}
158
159/// Whether the device sees monsters only as the players do (see
160/// [`edda_types::combat::view`]).
161fn behind_the_screen(claims: &Claims) -> bool {
162    claims.role == Principal::Player
163}
164
165/// `events`, which follow the combat as it stands in `state`, as the device
166/// sees them.
167fn seen(claims: &Claims, state: &CombatState, events: Vec<LoggedEvent>) -> Vec<LoggedEvent> {
168    if !behind_the_screen(claims) {
169        return events;
170    }
171    let mut concealer = Concealer::new(state);
172    events.iter().map(|e| concealer.event(e)).collect()
173}
174
175#[derive(Deserialize, utoipa::IntoParams)]
176pub(crate) struct LogQuery {
177    /// Only the events after this `seq`.
178    #[serde(default)]
179    after: u64,
180}
181
182#[utoipa::path(
183    get,
184    path = "/combats/{id}/log",
185    tag = "combats",
186    params(("id" = Uuid, Path, description = "The combat's id."), LogQuery),
187    responses(
188        (status = 200, description = "The events, in order. A player's device sees monsters' numbers concealed.", body = Vec<LoggedEvent>),
189        (status = 404, description = "No such combat in the campaign.", body = ErrorBody),
190    )
191)]
192/// The complete log, or the part after `?after=<seq>`.
193async fn log(
194    State(hub): State<AppState>,
195    Authenticated(claims): Authenticated,
196    Path(id): Path<Uuid>,
197    Query(query): Query<LogQuery>,
198) -> Result<Json<Vec<LoggedEvent>>, ApiError> {
199    hub.combat_in(id, claims.campaign).await?;
200    if !behind_the_screen(&claims) {
201        let events = hub.store().events(id, query.after).await?;
202        return Ok(Json(events.ok_or(ApiError::NotFound)?));
203    }
204    // Concealing needs to know everyone from the start.
205    let log = hub.store().events(id, 0).await?.ok_or(ApiError::NotFound)?;
206    let mut seen = seen(&claims, &CombatState::default(), log);
207    seen.retain(|e| e.seq > query.after);
208    Ok(Json(seen))
209}
210
211#[utoipa::path(
212    post,
213    path = "/combats/{id}/commands",
214    tag = "combats",
215    params(("id" = Uuid, Path, description = "The combat's id.")),
216    request_body = CommandEnvelope,
217    responses(
218        (status = 200, description = "The events it produced (the same ones again for a resent `command_id`).", body = Vec<LoggedEvent>),
219        (status = 403, description = "Not this device's to do, or the combat has as many events as it may keep.", body = ErrorBody),
220        (status = 404, description = "No such combat in the campaign.", body = ErrorBody),
221        (status = 422, description = "The rules refuse it, or the character has no slots or uses left for it.", body = ErrorBody),
222        (status = 502, description = "The character or monster service can't be reached.", body = ErrorBody),
223    )
224)]
225/// Runs a command and returns the events it produced. Resending a command
226/// returns its original events rather than applying it twice.
227async fn command(
228    State(hub): State<AppState>,
229    Authenticated(claims): Authenticated,
230    Path(id): Path<Uuid>,
231    Json(envelope): Json<CommandEnvelope>,
232) -> Result<Json<Vec<LoggedEvent>>, ApiError> {
233    let combat = hub.combat_in(id, claims.campaign).await?;
234    let envelope = attributed(envelope, &claims);
235    let command_id = envelope.command_id;
236    let events = match hub.execute(id, envelope, &claims).await? {
237        Outcome::Applied(events) => events,
238        Outcome::AlreadyApplied => hub.store().events_for_command(id, command_id).await?,
239    };
240    let state = combat.snapshot(hub.store()).await?.state;
241    Ok(Json(seen(&claims, &state, events)))
242}
243
244/// Live: the server sends a `ServerMessage` snapshot, then every event in
245/// log order; the client sends `ClientMessage`s (commands, undos). Both are
246/// JSON text messages.
247#[utoipa::path(
248    get,
249    path = "/combats/{id}/ws",
250    tag = "combats",
251    params(("id" = Uuid, Path, description = "The combat's id.")),
252    responses(
253        (status = 101, description = "Upgraded to a WebSocket."),
254        (status = 403, description = "As many devices are watching the combat as may.", body = ErrorBody),
255        (status = 404, description = "No such combat in the campaign.", body = ErrorBody),
256    )
257)]
258async fn websocket(
259    State(hub): State<AppState>,
260    Authenticated(claims): Authenticated,
261    Path(id): Path<Uuid>,
262    headers: HeaderMap,
263    upgrade: WebSocketUpgrade,
264) -> Result<Response, ApiError> {
265    // Resolve the combat first so an unknown id is a plain 404.
266    let combat = hub.combat_in(id, claims.campaign).await?;
267    let watching = i64::try_from(combat.watchers()).unwrap_or(i64::MAX);
268    let limits = hub.limits().with(&claims.settings.quotas);
269    limits
270        .room_for(
271            watching,
272            1,
273            limits.connections_per_combat,
274            "devices watching a combat",
275        )
276        .map_err(ApiError::Forbidden)?;
277    // Whose devices are connected, so an outage can be told to them.
278    let account = headers
279        .get(edda_server::observe::ACCOUNT)
280        .and_then(|v| v.to_str().ok())
281        .unwrap_or_default()
282        .to_owned();
283    Ok(upgrade.on_upgrade(move |socket| {
284        let watching = Watching::new(account);
285        async move {
286            session(socket, hub, combat, id, claims).await;
287            drop(watching);
288        }
289    }))
290}
291
292/// An open WebSocket, counted by account while it lasts:
293/// `edda_combat_watchers`.
294struct Watching(String);
295
296impl Watching {
297    fn new(account: String) -> Self {
298        metrics::gauge!("edda_combat_watchers", "account" => account.clone()).increment(1.0);
299        Self(account)
300    }
301}
302
303impl Drop for Watching {
304    fn drop(&mut self) {
305        metrics::gauge!("edda_combat_watchers", "account" => self.0.clone()).decrement(1.0);
306    }
307}
308
309/// The log records who sent a command as the authenticated device's name,
310/// whatever the client claimed.
311fn attributed(mut envelope: CommandEnvelope, claims: &Claims) -> CommandEnvelope {
312    envelope.actor = Some(claims.name.clone());
313    envelope
314}
315
316/// One connected client: snapshot first, then every event as it happens,
317/// while executing whatever commands the client sends.
318async fn session(
319    socket: WebSocket,
320    hub: AppState,
321    combat: Arc<LiveCombat>,
322    id: Uuid,
323    claims: Claims,
324) {
325    let (mut sink, mut incoming) = socket.split();
326    let Ok((snapshot, mut events)) = combat.subscribe(hub.store()).await else {
327        return;
328    };
329    // A player's device sees monsters as the players do.
330    let mut view = View::new(&claims);
331    if send(&mut sink, &ServerMessage::Snapshot(view.snapshot(snapshot)))
332        .await
333        .is_err()
334    {
335        return;
336    }
337    loop {
338        let reply = tokio::select! {
339            event = events.recv() => match event {
340                Ok(event) => ServerMessage::Event(view.event(&event)),
341                // Too far behind to catch up event by event: start over
342                // from a fresh snapshot.
343                Err(RecvError::Lagged(_)) => match combat.subscribe(hub.store()).await {
344                    Ok((snapshot, receiver)) => {
345                        events = receiver;
346                        ServerMessage::Snapshot(view.snapshot(snapshot))
347                    }
348                    Err(_) => break,
349                },
350                Err(RecvError::Closed) => break,
351            },
352            message = incoming.next() => match message {
353                Some(Ok(Message::Text(text))) => match handle(&hub, id, &text, &claims).await {
354                    Some(reply) => reply,
355                    None => continue,
356                },
357                Some(Ok(Message::Close(_)) | Err(_)) | None => break,
358                Some(Ok(_)) => continue,
359            },
360        };
361        if send(&mut sink, &reply).await.is_err() {
362            break;
363        }
364    }
365}
366
367/// What one connected device sees: everything, or for a player's device
368/// monsters concealed.
369struct View {
370    concealer: Option<Concealer>,
371    player: bool,
372}
373
374impl View {
375    fn new(claims: &Claims) -> Self {
376        Self {
377            concealer: None,
378            player: behind_the_screen(claims),
379        }
380    }
381
382    /// Starts over from `snapshot`.
383    fn snapshot(&mut self, mut snapshot: CombatSnapshot) -> CombatSnapshot {
384        if self.player {
385            let concealer = Concealer::new(&snapshot.state);
386            snapshot.state = concealer.state(&snapshot.state);
387            self.concealer = Some(concealer);
388        }
389        snapshot
390    }
391
392    fn event(&mut self, event: &LoggedEvent) -> LoggedEvent {
393        match &mut self.concealer {
394            Some(concealer) => concealer.event(event),
395            None => event.clone(),
396        }
397    }
398}
399
400/// Executes a client's command. Accepted commands need no direct reply:
401/// their events reach this client through the broadcast like everyone else.
402async fn handle(hub: &Hub, id: Uuid, text: &str, claims: &Claims) -> Option<ServerMessage> {
403    let ClientMessage::Command(envelope) = match serde_json::from_str(text) {
404        Ok(message) => message,
405        Err(e) => {
406            return Some(ServerMessage::Rejected {
407                command_id: Uuid::nil(),
408                error: format!("invalid message: {e}"),
409            });
410        }
411    };
412    let command_id = envelope.command_id;
413    match hub.execute(id, attributed(envelope, claims), claims).await {
414        Ok(Outcome::Applied(_)) => None,
415        Ok(Outcome::AlreadyApplied) => Some(ServerMessage::AlreadyApplied { command_id }),
416        Err(e) => Some(ServerMessage::Rejected {
417            command_id,
418            error: e.to_string(),
419        }),
420    }
421}
422
423async fn send(
424    sink: &mut (impl SinkExt<Message, Error = axum::Error> + Unpin),
425    message: &ServerMessage,
426) -> Result<(), axum::Error> {
427    let text = serde_json::to_string(message).expect("server messages always serialize");
428    sink.send(Message::Text(text.into())).await
429}
430
431#[cfg(test)]
432mod tests {
433    use metrics_exporter_prometheus::PrometheusBuilder;
434
435    use super::Watching;
436
437    #[test]
438    fn open_websockets_are_counted_by_account_while_they_last() {
439        let recorder = PrometheusBuilder::new().build_recorder();
440        let measured = recorder.handle();
441        metrics::with_local_recorder(&recorder, || {
442            let first = Watching::new("vox".into());
443            let second = Watching::new("vox".into());
444            assert!(
445                measured
446                    .render()
447                    .contains(r#"edda_combat_watchers{account="vox"} 2"#)
448            );
449            drop(first);
450            drop(second);
451            assert!(
452                measured
453                    .render()
454                    .contains(r#"edda_combat_watchers{account="vox"} 0"#)
455            );
456        });
457    }
458}
combat_service/routes.rs

combat_service/
routes.rs
