combat_service/
hub.rs

1//! Live combats on this instance: command execution and fan-out of events
2//! to connected clients.
3//!
4//! Nothing here is the source of truth; the database is. Several instances
5//! can serve the same combat at once:
6//!
7//! - Commands are serialised by locking the combat's row, and each instance
8//!   catches its cached state up from the log under that lock before
9//!   deciding anything.
10//! - Every append is announced with a Postgres notification. Each instance
11//!   listens, reads the new events from the log and pushes them to its own
12//!   clients, so a command executed anywhere reaches everyone.
13//!
14//! A combat nobody is watching is dropped from memory after a while and
15//! rebuilt from its log when next used, so combats can sit paused for days.
16
17use std::{
18    collections::{HashMap, HashSet},
19    sync::{Arc, Mutex as StdMutex},
20    time::{Duration, Instant},
21};
22
23use edda_auth::{
24    protocol::{Claims, Principal, Quotas},
25    server::READ_ONLY,
26};
27use edda_server::Limits;
28use edda_types::{
29    combat::{
30        CombatCommand, CombatEvent, CombatSnapshot, CombatState, CommandEnvelope, Dice,
31        LoggedEvent, SheetCost, add_monsters, decide, player_may, undo,
32    },
33    monster::Monster,
34};
35use sqlx::postgres::PgListener;
36use tokio::sync::{Mutex, broadcast};
37use uuid::Uuid;
38
39use crate::{
40    encounters,
41    error::ApiError,
42    services::Monsters,
43    store::{self, Notice, Store},
44    sync::SheetSync,
45};
46
47/// How many events a slow subscriber may fall behind before it is resynced
48/// with a fresh snapshot instead.
49const SUBSCRIBER_BUFFER: usize = 256;
50
51/// How long a combat with no connected clients stays in memory.
52const IDLE_TIMEOUT: Duration = Duration::from_mins(10);
53
54/// Every combat this instance has in memory, and what they need: the
55/// store, the other services and the limits.
56pub struct Hub {
57    store: Store,
58    live: Mutex<HashMap<Uuid, Arc<LiveCombat>>>,
59    sheet_sync: Option<Arc<SheetSync>>,
60    monsters: Option<Monsters>,
61    limits: Limits,
62}
63
64/// A combat in memory: its state, and the feed its watchers get.
65pub struct LiveCombat {
66    id: Uuid,
67    /// The campaign it belongs to; fixed at creation.
68    campaign: String,
69    /// The state commands are decided against. Held while a command runs.
70    core: Mutex<View>,
71    /// What this instance has broadcast so far. Subscribers start from a
72    /// snapshot of it, taken under the same lock, so they miss nothing.
73    feed: Mutex<View>,
74    events: broadcast::Sender<LoggedEvent>,
75    last_used: StdMutex<Instant>,
76}
77
78#[derive(Clone)]
79struct View {
80    state: CombatState,
81    seq: u64,
82}
83
84impl View {
85    fn apply(&mut self, event: &LoggedEvent) {
86        self.state.apply(&event.event);
87        self.seq = event.seq;
88    }
89}
90
91/// What running a command did.
92pub enum Outcome {
93    Applied(Vec<LoggedEvent>),
94    AlreadyApplied,
95}
96
97/// Fails unless a combat that has logged `seq` events may log `more`
98/// within `limits`.
99fn room_for_events(limits: &Limits, seq: u64, more: usize) -> Result<(), ApiError> {
100    let seq = i64::try_from(seq).unwrap_or(i64::MAX);
101    let more = u32::try_from(more).unwrap_or(u32::MAX);
102    limits
103        .room_for(
104            seq,
105            more,
106            limits.events_per_combat,
107            "events in a combat; start a new one",
108        )
109        .map_err(ApiError::Forbidden)
110}
111
112struct ServerDice;
113
114impl Dice for ServerDice {
115    fn roll(&mut self, sides: u16) -> Option<u16> {
116        Some(rand::random_range(1..=sides.max(1)))
117    }
118}
119
120impl Hub {
121    /// Starts listening on `listener` for events from all instances, and the background
122    /// upkeep: evicting idle combats and retrying HP write-back. Monsters
123    /// are added from `monsters`.
124    pub async fn start(
125        store: Store,
126        mut listener: PgListener,
127        sheet_sync: Option<SheetSync>,
128        monsters: Option<Monsters>,
129        limits: Limits,
130    ) -> Result<Arc<Self>, sqlx::Error> {
131        listener.listen(store::CHANNEL).await?;
132        let hub = Arc::new(Self {
133            store,
134            live: Mutex::new(HashMap::new()),
135            sheet_sync: sheet_sync.map(Arc::new),
136            monsters,
137            limits,
138        });
139        tokio::spawn(hub.clone().listen(listener));
140        tokio::spawn(hub.clone().evict_idle());
141        if let Some(sync) = &hub.sheet_sync {
142            tokio::spawn(sync.clone().retry_forever());
143        }
144        Ok(hub)
145    }
146
147    /// Where combats are kept.
148    pub fn store(&self) -> &Store {
149        &self.store
150    }
151
152    /// How much a campaign may keep.
153    pub fn limits(&self) -> &Limits {
154        &self.limits
155    }
156
157    /// Fails unless `campaign` may start another combat within `limits`.
158    async fn room_for_a_combat(&self, campaign: &str, limits: &Limits) -> Result<(), ApiError> {
159        let kept = self.store.count(campaign).await?;
160        limits
161            .room_for(kept, 1, limits.combats, "combats")
162            .map_err(ApiError::Forbidden)
163    }
164
165    /// Starts an empty combat in `campaign`, within its `quotas`.
166    pub async fn create(
167        &self,
168        campaign: &str,
169        name: &str,
170        quotas: &Quotas,
171    ) -> Result<CombatSnapshot, ApiError> {
172        self.room_for_a_combat(campaign, &self.limits.with(quotas))
173            .await?;
174        let id = self.store.create(campaign, name, Vec::new()).await?;
175        self.combat(id).await?.snapshot(&self.store).await
176    }
177
178    /// Creates a combat from one of `campaign`'s plans, with the plan's
179    /// monsters in it, and called `name` or after the plan. All or nothing:
180    /// if any of the monsters can't be added, there's no combat.
181    pub async fn create_from_plan(
182        &self,
183        campaign: Uuid,
184        encounter: Uuid,
185        name: Option<&str>,
186        quotas: &Quotas,
187    ) -> Result<CombatSnapshot, ApiError> {
188        let limits = self.limits.with(quotas);
189        self.room_for_a_combat(&campaign.to_string(), &limits)
190            .await?;
191        let plan = encounters::plan(self, campaign, encounter).await?;
192        let name = name.filter(|n| !n.trim().is_empty()).unwrap_or(&plan.name);
193        let mut state = CombatState::default();
194        state.apply(&CombatEvent::Created {
195            name: name.to_string(),
196            campaign: campaign.to_string(),
197        });
198        let mut setup = Vec::new();
199        for entry in &plan.monsters {
200            let monster = match self.stat_block(campaign, entry.monster).await {
201                Err(ApiError::BadRequest(_)) => {
202                    return Err(ApiError::BadRequest(format!(
203                        "{} from the plan isn't among the campaign's monsters any more",
204                        entry.name
205                    )));
206                }
207                other => other?,
208            };
209            for event in add_monsters(&state, &monster, entry.count, entry.roll, &mut ServerDice)? {
210                state.apply(&event);
211                setup.push(event);
212            }
213        }
214        room_for_events(&limits, 1, setup.len())?;
215        let id = self
216            .store
217            .create(&campaign.to_string(), name, setup)
218            .await?;
219        self.combat(id).await?.snapshot(&self.store).await
220    }
221
222    /// The characters a player's device plays, which are its combatants.
223    async fn played_by(
224        &self,
225        combat: &LiveCombat,
226        device: &str,
227    ) -> Result<HashSet<Uuid>, ApiError> {
228        let sync = self.sheet_sync.as_ref().ok_or_else(|| {
229            ApiError::Forbidden(
230                "players can't act here: this combat service can't reach the character sheets \
231                 to know whose they are"
232                    .into(),
233            )
234        })?;
235        sync.played_by(combat.campaign_id()?, device)
236            .await
237            .map_err(|e| ApiError::SheetUnavailable(e.to_string()))
238    }
239
240    /// A monster's stat block, from the monster service.
241    async fn stat_block(&self, campaign: Uuid, monster: Uuid) -> Result<Monster, ApiError> {
242        let monsters = self.monsters.as_ref().ok_or_else(|| {
243            ApiError::BadRequest(
244                "this combat service can't add monsters (no monster service)".into(),
245            )
246        })?;
247        monsters
248            .get(campaign, monster)
249            .await
250            .map_err(ApiError::from_monsters)
251    }
252
253    /// The live combat, loading it from its log if it isn't in memory.
254    pub async fn combat(&self, id: Uuid) -> Result<Arc<LiveCombat>, ApiError> {
255        // Held while loading, so a notification for this combat waits until
256        // it is in the map rather than being missed.
257        let mut live = self.live.lock().await;
258        if let Some(combat) = live.get(&id) {
259            combat.touch();
260            return Ok(combat.clone());
261        }
262        let log = self.store.events(id, 0).await?.ok_or(ApiError::NotFound)?;
263        let combat = Arc::new(LiveCombat::new(id, &log));
264        live.insert(id, combat.clone());
265        Ok(combat)
266    }
267
268    /// The combat, if it belongs to `campaign`. Another campaign's combat
269    /// is reported as not found, so its existence isn't revealed.
270    pub async fn combat_in(&self, id: Uuid, campaign: Uuid) -> Result<Arc<LiveCombat>, ApiError> {
271        let combat = self.combat(id).await?;
272        if combat.campaign != campaign.to_string() {
273            return Err(ApiError::NotFound);
274        }
275        Ok(combat)
276    }
277
278    /// Runs a command sent by the device `claims` are for. A player's
279    /// device may only change its own combatants (see [`player_may`]).
280    /// What the command costs a character's sheet is spent first (no slot
281    /// left, no cast), and given back if the command is then refused.
282    pub async fn execute(
283        &self,
284        id: Uuid,
285        envelope: CommandEnvelope,
286        claims: &Claims,
287    ) -> Result<Outcome, ApiError> {
288        // Commands over a WebSocket aren't requests of their own, so the
289        // token check didn't see them.
290        if claims.settings.read_only {
291            return Err(ApiError::Forbidden(READ_ONLY.into()));
292        }
293        let combat = self.combat(id).await?;
294        let command_id = envelope.command_id;
295        let actor_key = claims.key.to_string();
296        let limits = self.limits.with(&claims.settings.quotas);
297        // Checked before anything is spent, and again under the combat's
298        // lock against the latest state.
299        let played = match claims.role {
300            Principal::Player => {
301                let played = self.played_by(&combat, &actor_key).await?;
302                let state = combat.snapshot(&self.store).await?.state;
303                permit(&state, &envelope.command, Some(&played))?;
304                Some(played)
305            }
306            Principal::Dm | Principal::Service => None,
307        };
308        let stat_block = match &envelope.command {
309            CombatCommand::AddMonsters { monster, .. } => {
310                Some(self.stat_block(combat.campaign_id()?, *monster).await?)
311            }
312            _ => None,
313        };
314        let spent = self.spend(&combat, &envelope).await?;
315        let outcome = combat
316            .execute(
317                &self.store,
318                envelope,
319                &actor_key,
320                played.as_ref(),
321                stat_block.as_ref(),
322                |seq, more| room_for_events(&limits, seq, more),
323            )
324            .await;
325        if let (Err(_), Some((sync, campaign, character, cost))) = (&outcome, &spent) {
326            let key = crate::sync::refused_key(command_id);
327            if let Err(e) = sync.give_back(*campaign, *character, cost, key).await {
328                tracing::error!(%id, %character, error = %e, "giving back what a refused command spent failed");
329            }
330        }
331        let outcome = outcome?;
332        if let (Outcome::Applied(events), Some(sync)) = (&outcome, &self.sheet_sync) {
333            let undo = events
334                .iter()
335                .any(|e| matches!(e.event, CombatEvent::Undone { .. }));
336            if undo {
337                // What the undone command spent is back on the sheet before
338                // anyone is told it was undone.
339                sync.clone().run(id).await;
340            } else {
341                tokio::spawn(sync.clone().run(id));
342            }
343        }
344        Ok(outcome)
345    }
346
347    /// Spends what `envelope`'s command costs a character's sheet, if
348    /// anything; what to give back if the command is then refused.
349    async fn spend(
350        &self,
351        combat: &LiveCombat,
352        envelope: &CommandEnvelope,
353    ) -> Result<Option<(Arc<SheetSync>, Uuid, Uuid, SheetCost)>, ApiError> {
354        let (Some(sync), Some((combatant, cost))) =
355            (&self.sheet_sync, envelope.command.sheet_cost())
356        else {
357            return Ok(None);
358        };
359        let snapshot = combat.snapshot(&self.store).await?;
360        let Some(character) = snapshot
361            .state
362            .combatant(combatant)
363            .and_then(|c| c.character_id)
364        else {
365            return Ok(None);
366        };
367        let campaign = combat.campaign_id()?;
368        sync.spend(campaign, character, &cost, envelope.command_id)
369            .await
370            .map_err(ApiError::from_sheet)?;
371        Ok(Some((sync.clone(), campaign, character, cost)))
372    }
373
374    async fn listen(self: Arc<Self>, mut listener: PgListener) {
375        loop {
376            match listener.try_recv().await {
377                Ok(Some(notification)) => {
378                    match serde_json::from_str::<Notice>(notification.payload()) {
379                        Ok(notice) => {
380                            let combat = self.live.lock().await.get(&notice.combat).cloned();
381                            if let Some(combat) = combat {
382                                let store = self.store.clone();
383                                tokio::spawn(async move {
384                                    if let Err(e) = combat.catch_up(&store, Some(notice.seq)).await
385                                    {
386                                        tracing::error!(error = %e, "broadcasting events failed");
387                                    }
388                                });
389                            }
390                        }
391                        Err(e) => tracing::warn!(error = %e, "ignoring malformed notification"),
392                    }
393                }
394                // The connection dropped; the listener reconnects on the next
395                // call, but notifications sent meanwhile are lost.
396                Ok(None) => {
397                    tracing::warn!("lost connection for notifications; catching up");
398                    let combats: Vec<_> = self.live.lock().await.values().cloned().collect();
399                    for combat in combats {
400                        if let Err(e) = combat.catch_up(&self.store, None).await {
401                            tracing::error!(error = %e, "catching up failed");
402                        }
403                    }
404                }
405                Err(e) => {
406                    tracing::error!(error = %e, "listening for notifications failed");
407                    tokio::time::sleep(Duration::from_secs(1)).await;
408                }
409            }
410        }
411    }
412
413    async fn evict_idle(self: Arc<Self>) {
414        let mut interval = tokio::time::interval(Duration::from_mins(1));
415        loop {
416            interval.tick().await;
417            self.live
418                .lock()
419                .await
420                .retain(|_, c| c.events.receiver_count() > 0 || c.idle_for() < IDLE_TIMEOUT);
421        }
422    }
423}
424
425impl LiveCombat {
426    fn new(id: Uuid, log: &[LoggedEvent]) -> Self {
427        let view = View {
428            state: CombatState::replay(log.iter().map(|e| &e.event)),
429            seq: log.last().map_or(0, |e| e.seq),
430        };
431        Self {
432            id,
433            campaign: view.state.campaign.clone(),
434            core: Mutex::new(view.clone()),
435            feed: Mutex::new(view),
436            events: broadcast::channel(SUBSCRIBER_BUFFER).0,
437            last_used: StdMutex::new(Instant::now()),
438        }
439    }
440
441    fn campaign_id(&self) -> Result<Uuid, ApiError> {
442        self.campaign
443            .parse()
444            .map_err(|e: uuid::Error| ApiError::Corrupt(e.to_string()))
445    }
446
447    /// How many devices on this instance are watching it.
448    pub fn watchers(&self) -> usize {
449        self.events.receiver_count()
450    }
451
452    fn touch(&self) {
453        *self.last_used.lock().expect("not poisoned") = Instant::now();
454    }
455
456    fn idle_for(&self) -> Duration {
457        self.last_used.lock().expect("not poisoned").elapsed()
458    }
459
460    /// The latest state, including events appended by other instances.
461    pub async fn snapshot(&self, store: &Store) -> Result<CombatSnapshot, ApiError> {
462        self.catch_up(store, None).await?;
463        let feed = self.feed.lock().await;
464        Ok(self.snapshot_of(&feed))
465    }
466
467    /// A snapshot plus a receiver for every event after it.
468    pub async fn subscribe(
469        &self,
470        store: &Store,
471    ) -> Result<(CombatSnapshot, broadcast::Receiver<LoggedEvent>), ApiError> {
472        self.catch_up(store, None).await?;
473        let feed = self.feed.lock().await;
474        Ok((self.snapshot_of(&feed), self.events.subscribe()))
475    }
476
477    fn snapshot_of(&self, view: &View) -> CombatSnapshot {
478        CombatSnapshot {
479            id: self.id,
480            seq: view.seq,
481            state: view.state.clone(),
482        }
483    }
484
485    /// Reads events past what has been broadcast and broadcasts them. With
486    /// `Some(seq)`, does nothing if that position was already reached.
487    async fn catch_up(&self, store: &Store, seq: Option<u64>) -> Result<(), ApiError> {
488        let mut feed = self.feed.lock().await;
489        if seq.is_some_and(|seq| seq <= feed.seq) {
490            return Ok(());
491        }
492        // Gone from the log means deleted (with its campaign), whatever this
493        // instance still has in memory.
494        let events = store
495            .events(self.id, feed.seq)
496            .await?
497            .ok_or(ApiError::NotFound)?;
498        for event in events {
499            feed.apply(&event);
500            // No subscribers is fine; the log is the record.
501            let _ = self.events.send(event);
502        }
503        Ok(())
504    }
505
506    /// Runs a command. For a player's device, `played` are the characters
507    /// it plays; `AddMonsters` needs the monster's `stat_block`.
508    async fn execute(
509        &self,
510        store: &Store,
511        envelope: CommandEnvelope,
512        actor_key: &str,
513        played: Option<&HashSet<Uuid>>,
514        stat_block: Option<&Monster>,
515        room_for_events: impl Fn(u64, usize) -> Result<(), ApiError>,
516    ) -> Result<Outcome, ApiError> {
517        let mut core = self.core.lock().await;
518        let mut tx = store.begin().await?;
519        let latest = store
520            .lock(&mut tx, self.id)
521            .await?
522            .ok_or(ApiError::NotFound)?;
523        if latest > core.seq {
524            // Other instances have appended since; decide against the
525            // state as it really is.
526            for event in store.events_in(&mut tx, self.id, core.seq).await? {
527                core.apply(&event);
528            }
529        }
530        permit(&core.state, &envelope.command, played)?;
531        if !store
532            .record_command(&mut tx, self.id, envelope.command_id, actor_key)
533            .await?
534        {
535            return Ok(Outcome::AlreadyApplied);
536        }
537        let events = match envelope.command {
538            CombatCommand::Undo { command } => {
539                let log = store.events_in(&mut tx, self.id, 0).await?;
540                let mine = store.commands_by(&mut tx, self.id, actor_key).await?;
541                undo::undo(&log, command, |c| mine.contains(&c))?
542            }
543            CombatCommand::AddMonsters { count, roll, .. } => {
544                let monster = stat_block.expect("fetched by Hub::execute");
545                add_monsters(&core.state, monster, count, roll, &mut ServerDice)?
546            }
547            ref command => decide(&core.state, command, &mut ServerDice)?,
548        };
549
550        room_for_events(core.seq, events.len())?;
551        let at = store::now();
552        let logged: Vec<_> = events
553            .into_iter()
554            .zip(core.seq + 1..)
555            .map(|(event, seq)| LoggedEvent {
556                seq,
557                command_id: Some(envelope.command_id),
558                actor: envelope.actor.clone(),
559                at,
560                event,
561            })
562            .collect();
563        let mut next = core.clone();
564        for event in &logged {
565            next.apply(event);
566        }
567        // Clients hear about the events through the notification sent on
568        // commit, the same way as about everyone else's.
569        store.append(&mut tx, self.id, &logged, &next.state).await?;
570        tx.commit().await?;
571        *core = next;
572        Ok(Outcome::Applied(logged))
573    }
574}
575
576/// Whether the device may send `command`: anything for a DM device or a
577/// service (`played` is `None`), only what [`player_may`] allows for a
578/// player's device, which plays the characters `played`.
579fn permit(
580    state: &CombatState,
581    command: &CombatCommand,
582    played: Option<&HashSet<Uuid>>,
583) -> Result<(), ApiError> {
584    let Some(played) = played else {
585        return Ok(());
586    };
587    player_may(state, command, |c| {
588        c.character_id.is_some_and(|id| played.contains(&id))
589    })
590    .map_err(|e| ApiError::Forbidden(e.to_string()))
591}
combat_service/hub.rs

combat_service/
hub.rs
