combat_service/
store.rs

1//! Persistence of the combat event log in Postgres, shared by every
2//! instance of the service.
3
4use std::collections::HashSet;
5
6use chrono::{DateTime, SubsecRound, Utc};
7use edda_types::combat::{CombatEvent, CombatState, CombatStatus, CombatSummary, LoggedEvent};
8use serde::{Deserialize, Serialize};
9use sqlx::{PgExecutor, PgPool, Postgres, Transaction, types::Json};
10use uuid::Uuid;
11
12use crate::error::ApiError;
13
14/// The Postgres channel on which every appended event is announced, so all
15/// instances can push it to their connected clients.
16pub const CHANNEL: &str = "combat_events";
17
18/// A transaction on the combats' database.
19pub type Tx = Transaction<'static, Postgres>;
20
21type SummaryRow = (
22    Uuid,
23    String,
24    String,
25    String,
26    i32,
27    DateTime<Utc>,
28    DateTime<Utc>,
29);
30
31type EventRow = (
32    i64,
33    Option<Uuid>,
34    Option<String>,
35    DateTime<Utc>,
36    Json<CombatEvent>,
37);
38
39/// Payload of a notification on [`CHANNEL`]: the combat's log now reaches
40/// `seq`. Events themselves are read from the table, which keeps
41/// notifications small and lets a listener that missed some catch up.
42#[derive(Debug, Serialize, Deserialize)]
43pub struct Notice {
44    pub combat: Uuid,
45    pub seq: u64,
46}
47
48/// Which combats to list.
49#[derive(Debug, Default, Deserialize)]
50pub struct ListFilter {
51    pub campaign: Option<String>,
52    pub status: Option<CombatStatus>,
53}
54
55/// Combats, their logs and planned encounters, in Postgres.
56#[derive(Clone)]
57pub struct Store {
58    pool: PgPool,
59}
60
61impl Store {
62    /// A store in `pool`'s database.
63    pub fn new(pool: PgPool) -> Self {
64        Self { pool }
65    }
66
67    /// The database.
68    pub fn pool(&self) -> &PgPool {
69        &self.pool
70    }
71
72    /// Starts a transaction.
73    pub async fn begin(&self) -> Result<Tx, ApiError> {
74        Ok(self.pool.begin().await?)
75    }
76
77    /// Creates a combat whose log starts with a `Created` event, followed
78    /// by `setup` (a plan's monsters, say), all at once.
79    pub async fn create(
80        &self,
81        campaign: &str,
82        name: &str,
83        setup: Vec<CombatEvent>,
84    ) -> Result<Uuid, ApiError> {
85        let id = Uuid::new_v4();
86        let at = now();
87        let created = CombatEvent::Created {
88            name: name.to_string(),
89            campaign: campaign.to_string(),
90        };
91        let log: Vec<LoggedEvent> = std::iter::once(created)
92            .chain(setup)
93            .zip(1..)
94            .map(|(event, seq)| LoggedEvent {
95                seq,
96                command_id: None,
97                actor: None,
98                at,
99                event,
100            })
101            .collect();
102        let seq = from_seq(log.len() as u64)?;
103        let mut tx = self.begin().await?;
104        sqlx::query(
105            "INSERT INTO combats (id, campaign, name, status, round, seq, created_at, updated_at) \
106             VALUES ($1, $2, $3, $4, 0, $5, $6, $6)",
107        )
108        .bind(id)
109        .bind(campaign)
110        .bind(name)
111        .bind(status_str(CombatStatus::Preparing))
112        .bind(seq)
113        .bind(at)
114        .execute(&mut *tx)
115        .await?;
116        for event in &log {
117            insert_event(&mut tx, id, event).await?;
118        }
119        sqlx::query("INSERT INTO combat_hp_sync (combat_id, synced_seq) VALUES ($1, $2)")
120            .bind(id)
121            .bind(seq)
122            .execute(&mut *tx)
123            .await?;
124        notify(&mut tx, id, log.len() as u64).await?;
125        tx.commit().await?;
126        Ok(id)
127    }
128
129    /// How many combats `campaign` keeps.
130    pub async fn count(&self, campaign: &str) -> Result<i64, ApiError> {
131        let (count,): (i64,) = sqlx::query_as("SELECT COUNT(*) FROM combats WHERE campaign = $1")
132            .bind(campaign)
133            .fetch_one(&self.pool)
134            .await?;
135        Ok(count)
136    }
137
138    /// Most recently active first.
139    pub async fn list(&self, filter: &ListFilter) -> Result<Vec<CombatSummary>, ApiError> {
140        let rows: Vec<SummaryRow> = sqlx::query_as(
141            "SELECT id, campaign, name, status, round, created_at, updated_at FROM combats \
142                 WHERE ($1::text IS NULL OR campaign = $1) AND ($2::text IS NULL OR status = $2) \
143                 ORDER BY updated_at DESC",
144        )
145        .bind(&filter.campaign)
146        .bind(filter.status.map(status_str))
147        .fetch_all(&self.pool)
148        .await?;
149        rows.into_iter()
150            .map(
151                |(id, campaign, name, status, round, created_at, updated_at)| {
152                    Ok(CombatSummary {
153                        id,
154                        campaign,
155                        name,
156                        status: serde_json::from_value(serde_json::Value::String(status))
157                            .map_err(|e| ApiError::Corrupt(e.to_string()))?,
158                        round: u32::try_from(round)
159                            .map_err(|e| ApiError::Corrupt(e.to_string()))?,
160                        created_at,
161                        updated_at,
162                    })
163                },
164            )
165            .collect()
166    }
167
168    /// The combat's log from `after + 1` onwards, or `None` if there's no
169    /// such combat.
170    pub async fn events(&self, id: Uuid, after: u64) -> Result<Option<Vec<LoggedEvent>>, ApiError> {
171        let exists: Option<(i32,)> = sqlx::query_as("SELECT 1 FROM combats WHERE id = $1")
172            .bind(id)
173            .fetch_optional(&self.pool)
174            .await?;
175        if exists.is_none() {
176            return Ok(None);
177        }
178        fetch_events(&self.pool, id, after).await.map(Some)
179    }
180
181    /// Deletes every combat of a campaign, logs and all; returns how many
182    /// there were.
183    /// (Its planned encounters go too.)
184    pub async fn delete_campaign(&self, campaign: &str) -> Result<u64, ApiError> {
185        sqlx::query("DELETE FROM encounters WHERE campaign = $1")
186            .bind(campaign)
187            .execute(&self.pool)
188            .await?;
189        // Events, commands and sync progress go with them (ON DELETE CASCADE).
190        let result = sqlx::query("DELETE FROM combats WHERE campaign = $1")
191            .bind(campaign)
192            .execute(&self.pool)
193            .await?;
194        Ok(result.rows_affected())
195    }
196
197    /// The events a given command produced.
198    pub async fn events_for_command(
199        &self,
200        id: Uuid,
201        command_id: Uuid,
202    ) -> Result<Vec<LoggedEvent>, ApiError> {
203        let rows: Vec<EventRow> = sqlx::query_as(
204            "SELECT seq, command_id, actor, at, event FROM combat_events \
205             WHERE combat_id = $1 AND command_id = $2 ORDER BY seq",
206        )
207        .bind(id)
208        .bind(command_id)
209        .fetch_all(&self.pool)
210        .await?;
211        rows.into_iter().map(from_row).collect()
212    }
213
214    /// Locks the combat until `tx` ends, so no other command (on any
215    /// instance) can append in the meantime. Returns the latest seq, or
216    /// `None` if there's no such combat.
217    pub async fn lock(&self, tx: &mut Tx, id: Uuid) -> Result<Option<u64>, ApiError> {
218        let row: Option<(i64,)> =
219            sqlx::query_as("SELECT seq FROM combats WHERE id = $1 FOR UPDATE")
220                .bind(id)
221                .fetch_optional(&mut **tx)
222                .await?;
223        row.map(|(seq,)| to_seq(seq)).transpose()
224    }
225
226    /// A combat's events after `after`, inside `tx`.
227    pub async fn events_in(
228        &self,
229        tx: &mut Tx,
230        id: Uuid,
231        after: u64,
232    ) -> Result<Vec<LoggedEvent>, ApiError> {
233        fetch_events(&mut **tx, id, after).await
234    }
235
236    /// Records that a command sent by the device with `actor_key` is being
237    /// applied. `false` if it already was.
238    pub async fn record_command(
239        &self,
240        tx: &mut Tx,
241        id: Uuid,
242        command_id: Uuid,
243        actor_key: &str,
244    ) -> Result<bool, ApiError> {
245        let inserted = sqlx::query(
246            "INSERT INTO combat_commands (combat_id, command_id, actor_key) VALUES ($1, $2, $3) \
247             ON CONFLICT DO NOTHING",
248        )
249        .bind(id)
250        .bind(command_id)
251        .bind(actor_key)
252        .execute(&mut **tx)
253        .await?;
254        Ok(inserted.rows_affected() == 1)
255    }
256
257    /// The commands to this combat sent by the device with `actor_key`.
258    pub async fn commands_by(
259        &self,
260        tx: &mut Tx,
261        id: Uuid,
262        actor_key: &str,
263    ) -> Result<HashSet<Uuid>, ApiError> {
264        let rows: Vec<(Uuid,)> = sqlx::query_as(
265            "SELECT command_id FROM combat_commands WHERE combat_id = $1 AND actor_key = $2",
266        )
267        .bind(id)
268        .bind(actor_key)
269        .fetch_all(&mut **tx)
270        .await?;
271        Ok(rows.into_iter().map(|(command,)| command).collect())
272    }
273
274    /// Appends events (non-empty, continuing the log), refreshes the
275    /// summary and announces them on [`CHANNEL`] once `tx` commits.
276    pub async fn append(
277        &self,
278        tx: &mut Tx,
279        id: Uuid,
280        events: &[LoggedEvent],
281        state: &CombatState,
282    ) -> Result<(), ApiError> {
283        let Some(last) = events.last() else {
284            return Ok(());
285        };
286        for event in events {
287            insert_event(tx, id, event).await?;
288        }
289        sqlx::query(
290            "UPDATE combats SET status = $1, round = $2, seq = $3, updated_at = $4 WHERE id = $5",
291        )
292        .bind(status_str(state.status))
293        .bind(i32::try_from(state.round).map_err(|e| ApiError::Corrupt(e.to_string()))?)
294        .bind(from_seq(last.seq)?)
295        .bind(last.at)
296        .bind(id)
297        .execute(&mut **tx)
298        .await?;
299        notify(tx, id, last.seq).await
300    }
301}
302
303/// The current time at the precision Postgres stores, so an event read back
304/// from the log equals the one that was written.
305pub fn now() -> DateTime<Utc> {
306    Utc::now().trunc_subsecs(6)
307}
308
309async fn fetch_events(
310    executor: impl PgExecutor<'_>,
311    id: Uuid,
312    after: u64,
313) -> Result<Vec<LoggedEvent>, ApiError> {
314    let rows: Vec<EventRow> = sqlx::query_as(
315        "SELECT seq, command_id, actor, at, event FROM combat_events \
316         WHERE combat_id = $1 AND seq > $2 ORDER BY seq",
317    )
318    .bind(id)
319    .bind(i64::try_from(after).unwrap_or(i64::MAX))
320    .fetch_all(executor)
321    .await?;
322    rows.into_iter().map(from_row).collect()
323}
324
325pub(crate) async fn insert_event(
326    tx: &mut Tx,
327    id: Uuid,
328    event: &LoggedEvent,
329) -> Result<(), ApiError> {
330    sqlx::query(
331        "INSERT INTO combat_events (combat_id, seq, command_id, actor, at, event) \
332         VALUES ($1, $2, $3, $4, $5, $6)",
333    )
334    .bind(id)
335    .bind(from_seq(event.seq)?)
336    .bind(event.command_id)
337    .bind(&event.actor)
338    .bind(event.at)
339    .bind(Json(&event.event))
340    .execute(&mut **tx)
341    .await?;
342    Ok(())
343}
344
345/// Postgres delivers the notification only if and when `tx` commits.
346async fn notify(tx: &mut Tx, id: Uuid, seq: u64) -> Result<(), ApiError> {
347    let payload = serde_json::to_string(&Notice { combat: id, seq })
348        .map_err(|e| ApiError::Corrupt(e.to_string()))?;
349    sqlx::query("SELECT pg_notify($1, $2)")
350        .bind(CHANNEL)
351        .bind(payload)
352        .execute(&mut **tx)
353        .await?;
354    Ok(())
355}
356
357fn from_row((seq, command_id, actor, at, Json(event)): EventRow) -> Result<LoggedEvent, ApiError> {
358    Ok(LoggedEvent {
359        seq: to_seq(seq)?,
360        command_id,
361        actor,
362        at,
363        event,
364    })
365}
366
367pub(crate) fn to_seq(seq: i64) -> Result<u64, ApiError> {
368    u64::try_from(seq).map_err(|e| ApiError::Corrupt(e.to_string()))
369}
370
371pub(crate) fn from_seq(seq: u64) -> Result<i64, ApiError> {
372    i64::try_from(seq).map_err(|e| ApiError::Corrupt(e.to_string()))
373}
374
375pub(crate) fn status_str(status: CombatStatus) -> &'static str {
376    match status {
377        CombatStatus::Preparing => "preparing",
378        CombatStatus::Active => "active",
379        CombatStatus::Paused => "paused",
380        CombatStatus::Ended => "ended",
381    }
382}
combat_service/store.rs

combat_service/
store.rs
