combat_service/
sync.rs

1//! Keeps character sheets in step with combats: hit points are written
2//! back, and what casting or using a feature costs (a slot, uses) is spent
3//! on the sheet before the command runs and given back if it's refused or
4//! undone. Every frontend gets this without doing anything.
5//!
6//! The combat log stays the authority during a fight; the character service
7//! follows it. How far each combat has been written back is stored in
8//! `combat_hp_sync`, so a write-back that fails (the character service is
9//! down, say) is retried later by whichever instance gets there first.
10//! Only one instance syncs a given combat at a time, and it always writes
11//! the latest HP, so sheets never go backwards. Spending and giving back
12//! carry an idempotency key, so a retry never does either twice.
13//!
14//! It writes as this service itself (see [`crate::services`]).
15
16use std::{
17    collections::{HashMap, HashSet},
18    sync::Arc,
19    time::Duration,
20};
21
22use edda_client::{Client, ClientError};
23use edda_types::{
24    HitPoints,
25    combat::{CombatEvent, CombatState, LoggedEvent, SheetCost},
26};
27use uuid::Uuid;
28
29use crate::{error::ApiError, services::ServiceLogin, store::Store};
30
31/// How often combats whose write-back failed are retried.
32const RETRY_INTERVAL: Duration = Duration::from_secs(30);
33
34/// Keeps character sheets in step with combats: hit points, and slots and
35/// uses spent or given back.
36pub struct SheetSync {
37    store: Store,
38    login: Arc<ServiceLogin>,
39    character_service: String,
40}
41
42/// Idempotency keys: what a command spends is keyed by the command itself;
43/// giving it back has a key of its own for each reason.
44fn give_back_key(command: Uuid, why: &str) -> Uuid {
45    Uuid::new_v5(&command, why.as_bytes())
46}
47
48/// Giving back what a refused command spent.
49pub fn refused_key(command: Uuid) -> Uuid {
50    give_back_key(command, "refused")
51}
52
53fn undone_key(command: Uuid) -> Uuid {
54    give_back_key(command, "undone")
55}
56
57#[derive(Debug, thiserror::Error)]
58enum SyncError {
59    #[error(transparent)]
60    Store(#[from] ApiError),
61    #[error("character service: {0}")]
62    Characters(#[from] ClientError),
63}
64
65impl SheetSync {
66    /// Writes to the character service at `character_service`, logged in
67    /// to with `login`.
68    pub fn new(store: Store, login: Arc<ServiceLogin>, character_service: String) -> Self {
69        Self {
70            store,
71            login,
72            character_service,
73        }
74    }
75
76    /// A character service client authorised for `campaign`.
77    async fn characters(&self, campaign: Uuid) -> Result<Client, ClientError> {
78        let token = self.login.token(campaign).await?;
79        Ok(Client::new(&self.character_service).with_token(&token))
80    }
81
82    /// The characters of `campaign` that the player device `device` (a key)
83    /// plays.
84    pub async fn played_by(
85        &self,
86        campaign: Uuid,
87        device: &str,
88    ) -> Result<HashSet<Uuid>, ClientError> {
89        let characters = self.characters(campaign).await?.list().await?;
90        Ok(characters
91            .into_iter()
92            .filter(|c| c.owners.iter().any(|owner| owner == device))
93            .map(|c| c.id)
94            .collect())
95    }
96
97    /// Spends `cost` on `character`'s sheet for `command`, once however
98    /// often the command is sent.
99    pub async fn spend(
100        &self,
101        campaign: Uuid,
102        character: Uuid,
103        cost: &SheetCost,
104        command: Uuid,
105    ) -> Result<(), ClientError> {
106        let characters = self.characters(campaign).await?;
107        characters
108            .act_once(character, &cost.spend(), command)
109            .await
110            .map(drop)
111    }
112
113    /// Gives `cost` back to `character`'s sheet, once per `key`.
114    pub async fn give_back(
115        &self,
116        campaign: Uuid,
117        character: Uuid,
118        cost: &SheetCost,
119        key: Uuid,
120    ) -> Result<(), ClientError> {
121        let characters = self.characters(campaign).await?;
122        match characters.act_once(character, &cost.give_back(), key).await {
123            Ok(_) | Err(ClientError::NotFound) => Ok(()),
124            Err(e) => Err(e),
125        }
126    }
127
128    /// Syncs `combat` until it is up to date, unless another instance is
129    /// already on it (that instance then picks up anything new too).
130    pub async fn run(self: Arc<Self>, combat: Uuid) {
131        loop {
132            match self.sync_once(combat).await {
133                Ok(true) => {}
134                Ok(false) => break,
135                Err(e) => {
136                    tracing::warn!(%combat, error = %e, "hit point write-back failed; will retry");
137                    break;
138                }
139            }
140        }
141    }
142
143    /// Retries failed write-backs every 30 seconds, for as long as the
144    /// service runs.
145    pub async fn retry_forever(self: Arc<Self>) {
146        let mut interval = tokio::time::interval(RETRY_INTERVAL);
147        loop {
148            interval.tick().await;
149            let behind: Result<Vec<(Uuid,)>, _> = sqlx::query_as(
150                "SELECT s.combat_id FROM combat_hp_sync s JOIN combats c ON c.id = s.combat_id \
151                 WHERE s.synced_seq < c.seq",
152            )
153            .fetch_all(self.store.pool())
154            .await;
155            match behind {
156                Ok(combats) => {
157                    metrics::gauge!("edda_combat_hp_sync_behind")
158                        .set(edda_server::measure::as_f64(combats.len()));
159                    for (combat,) in combats {
160                        self.clone().run(combat).await;
161                    }
162                }
163                Err(e) => tracing::warn!(error = %e, "finding combats to sync failed"),
164            }
165        }
166    }
167
168    /// One pass: writes back every HP change since the last sync. `true` if
169    /// it made progress (so there may be more to do), `false` if there was
170    /// nothing to do or another instance holds the combat.
171    async fn sync_once(&self, combat: Uuid) -> Result<bool, SyncError> {
172        let mut tx = self.store.begin().await?;
173        let row: Option<(i64,)> = sqlx::query_as(
174            "SELECT synced_seq FROM combat_hp_sync WHERE combat_id = $1 FOR UPDATE SKIP LOCKED",
175        )
176        .bind(combat)
177        .fetch_optional(&mut *tx)
178        .await
179        .map_err(ApiError::from)?;
180        let Some((synced,)) = row else {
181            return Ok(false);
182        };
183        let synced = u64::try_from(synced).unwrap_or(0);
184        let log = self.store.events_in(&mut tx, combat, 0).await?;
185        let Some(latest) = log.last().map(|e| e.seq).filter(|&seq| seq > synced) else {
186            return Ok(false);
187        };
188
189        let changes = hp_changes(&log, synced);
190        let refunds = undone_costs(&log, synced);
191        let campaign = CombatState::replay(log.iter().map(|e| &e.event)).campaign;
192        if !changes.is_empty() || !refunds.is_empty() {
193            match campaign.parse() {
194                Ok(campaign) => {
195                    self.write(combat, campaign, &changes).await?;
196                    for (character, cost, command) in &refunds {
197                        self.give_back(campaign, *character, cost, undone_key(*command))
198                            .await?;
199                    }
200                }
201                Err(_) => {
202                    tracing::warn!(%combat, %campaign, "combat has no valid campaign; not syncing");
203                }
204            }
205        }
206
207        sqlx::query("UPDATE combat_hp_sync SET synced_seq = $1 WHERE combat_id = $2")
208            .bind(i64::try_from(latest).unwrap_or(i64::MAX))
209            .bind(combat)
210            .execute(&mut *tx)
211            .await
212            .map_err(ApiError::from)?;
213        tx.commit().await.map_err(ApiError::from)?;
214        Ok(true)
215    }
216
217    async fn write(
218        &self,
219        combat: Uuid,
220        campaign: Uuid,
221        changes: &HashMap<Uuid, HitPoints>,
222    ) -> Result<(), SyncError> {
223        let characters = self.characters(campaign).await?;
224        for (&character, &hp) in changes {
225            let written = characters
226                .modify(character, |sheet| {
227                    let sheet_hp = &mut sheet.hit_points;
228                    // The sheet's maximum may have changed since the
229                    // character joined the combat.
230                    sheet_hp.current = hp.current.min(sheet_hp.max);
231                    sheet_hp.temporary = hp.temporary;
232                    Ok::<_, ClientError>(())
233                })
234                .await;
235            match written {
236                Ok(_) => {}
237                Err(ClientError::NotFound) => {
238                    tracing::info!(%combat, %character, "character no longer exists; skipping");
239                }
240                Err(e) => return Err(e.into()),
241            }
242        }
243        Ok(())
244    }
245}
246
247/// The latest hit points of every character whose HP changed after log
248/// position `after`, keyed by character id. Joining a combat isn't a
249/// change: a combatant starts with the sheet's own HP.
250fn hp_changes(log: &[LoggedEvent], after: u64) -> HashMap<Uuid, HitPoints> {
251    let mut state = CombatState::default();
252    let mut known: HashMap<Uuid, HitPoints> = HashMap::new();
253    let mut changes = HashMap::new();
254    for event in log {
255        state.apply(&event.event);
256        for combatant in &state.combatants {
257            let Some(character) = combatant.character_id else {
258                continue;
259            };
260            let previous = known.insert(combatant.id, combatant.hit_points);
261            if event.seq > after && previous.is_some_and(|hp| hp != combatant.hit_points) {
262                changes.insert(character, combatant.hit_points);
263            }
264        }
265    }
266    changes
267}
268
269/// What the commands undone after log position `after` had spent on
270/// characters' sheets: the character, the cost, and the undone command.
271fn undone_costs(log: &[LoggedEvent], after: u64) -> Vec<(Uuid, SheetCost, Uuid)> {
272    // Combatants may have left by the time their command is undone.
273    let mut characters: HashMap<Uuid, Uuid> = HashMap::new();
274    for event in log {
275        if let CombatEvent::CombatantAdded { combatant }
276        | CombatEvent::CombatantRestored { combatant } = &event.event
277            && let Some(character) = combatant.character_id
278        {
279            characters.insert(combatant.id, character);
280        }
281    }
282    log.iter()
283        .filter(|e| e.seq > after)
284        .filter_map(|e| match e.event {
285            CombatEvent::Undone { command } => Some(command),
286            _ => None,
287        })
288        .flat_map(|command| {
289            log.iter()
290                .filter(move |e| e.command_id == Some(command))
291                .filter_map(|e| e.event.sheet_cost())
292                .filter_map(|(combatant, cost)| {
293                    characters.get(&combatant).map(|c| (*c, cost, command))
294                })
295                .collect::<Vec<_>>()
296        })
297        .collect()
298}
299
300#[cfg(test)]
301mod tests {
302    use chrono::Utc;
303    use edda_types::combat::{CombatEvent, Combatant};
304
305    use super::*;
306
307    fn logged(seq: u64, event: CombatEvent) -> LoggedEvent {
308        LoggedEvent {
309            seq,
310            command_id: None,
311            actor: None,
312            at: Utc::now(),
313            event,
314        }
315    }
316
317    fn hp(current: u16, temporary: u16) -> HitPoints {
318        HitPoints {
319            max: 30,
320            current,
321            temporary,
322        }
323    }
324
325    #[test]
326    fn only_changes_after_the_sync_point_count_and_the_latest_wins() {
327        let character = Uuid::new_v4();
328        let mut vex = Combatant::new("Vex", 30);
329        vex.character_id = Some(character);
330        let goblin = Combatant::new("Goblin", 7);
331        let damaged = |seq, combatant, current| {
332            logged(
333                seq,
334                CombatEvent::Damaged {
335                    combatant,
336                    amount: 1,
337                    hit_points: hp(current, 0),
338                },
339            )
340        };
341        let log = vec![
342            logged(
343                1,
344                CombatEvent::Created {
345                    name: "A".into(),
346                    campaign: "C".into(),
347                },
348            ),
349            logged(
350                2,
351                CombatEvent::CombatantAdded {
352                    combatant: vex.clone(),
353                },
354            ),
355            logged(
356                3,
357                CombatEvent::CombatantAdded {
358                    combatant: goblin.clone(),
359                },
360            ),
361            damaged(4, vex.id, 25),
362            damaged(5, goblin.id, 2),
363            damaged(6, vex.id, 20),
364        ];
365        assert_eq!(hp_changes(&log, 3).len(), 1);
366        assert_eq!(hp_changes(&log, 3)[&character], hp(20, 0));
367        assert_eq!(hp_changes(&log, 5)[&character], hp(20, 0));
368        assert!(hp_changes(&log, 6).is_empty());
369        assert!(
370            hp_changes(&log[..3], 0).is_empty(),
371            "joining isn't a change"
372        );
373
374        // A character who leaves keeps the damage taken before leaving.
375        let mut log = log;
376        log.push(damaged(7, vex.id, 10));
377        log.push(logged(
378            8,
379            CombatEvent::CombatantRemoved { combatant: vex.id },
380        ));
381        assert_eq!(hp_changes(&log, 6)[&character], hp(10, 0));
382    }
383}
combat_service/sync.rs

combat_service/
sync.rs
