sofie_db/
dead_messages.rs

1use sofie_core::{Cast, ColumnReader, Error, Result, SqlValue, Timestamp, Uuid};
2
3use crate::{Db, Step, query_decoded};
4
5#[derive(Clone, Debug, PartialEq, Eq)]
6pub struct DeadMessage {
7    pub id: Uuid,
8    pub name: String,
9    pub attempts: i64,
10    pub last_error: Option<String>,
11    pub dead_at: Timestamp,
12}
13
14struct Stored {
15    id: Uuid,
16    name: String,
17    attempts: i64,
18    last_error: Option<String>,
19    dead_at: Option<Timestamp>,
20}
21
22impl Stored {
23    fn decode(row: &mut dyn ColumnReader) -> Result<Self> {
24        Ok(Self {
25            id: row.uuid()?,
26            name: row.text()?.to_owned(),
27            attempts: row.int()?,
28            last_error: Option::decode(row)?,
29            dead_at: Option::decode(row)?,
30        })
31    }
32
33    fn dead(self) -> Option<DeadMessage> {
34        Some(DeadMessage {
35            dead_at: self.dead_at?,
36            id: self.id,
37            name: self.name,
38            attempts: self.attempts,
39            last_error: self.last_error,
40        })
41    }
42}
43
44const COLUMNS: &str = "id, name, attempts, last_error, dead_at";
45
46pub struct DeadMessages {
47    db: Db,
48}
49
50impl DeadMessages {
51    #[must_use]
52    pub fn new(db: Db) -> Self {
53        Self { db }
54    }
55
56    pub async fn newest(&self, limit: u64) -> Result<Vec<DeadMessage>> {
57        let dialect = self.db.driver().dialect();
58        let sql = format!(
59            "SELECT {COLUMNS} FROM {} WHERE dead_at IS NOT NULL ORDER BY dead_at DESC LIMIT {}",
60            dialect.qualified("sofie", "messages"),
61            dialect.placeholder(1),
62        );
63        let limit = i64::try_from(limit)
64            .map_err(|error| Error::internal("a limit too large to bind", error))?;
65        let params = [SqlValue::Int(limit)];
66        let stored = query_decoded(self.db.driver(), &sql, &params, Vec::new(), Stored::decode);
67        Ok(stored.await?.into_iter().filter_map(Stored::dead).collect())
68    }
69
70    /// Logged at info as its audit, as no principal replays it.
71    pub async fn replay(&self, id: Uuid) -> Result<DeadMessage> {
72        let dead = self.dead_one(id).await?;
73        let dialect = self.db.driver().dialect();
74        let sql = format!(
75            "UPDATE {} SET attempts = 0, dead_at = NULL, claimed_until = NULL, available_at = {} \
76             WHERE id = {} AND attempts = {} AND dead_at IS NOT NULL",
77            dialect.qualified("sofie", "messages"),
78            dialect.placeholder(1),
79            dialect.placeholder(2),
80            dialect.placeholder(3),
81        );
82        let params = vec![
83            SqlValue::Timestamp(self.db.clock().now().timestamp()),
84            SqlValue::Uuid(id),
85            SqlValue::Int(dead.attempts),
86        ];
87        match self
88            .db
89            .transaction(vec![Step::statement(sql, params, Some(1))])
90            .await
91        {
92            Err(Error::Conflict) => Err(Error::Rule(format!(
93                "message {id} was replayed or changed meanwhile; nothing was replayed"
94            ))),
95            outcome => outcome,
96        }?;
97        tracing::info!(
98            message.id = %id,
99            message.name = %dead.name,
100            message.previous_attempts = dead.attempts,
101            "replayed a dead message"
102        );
103        Ok(dead)
104    }
105
106    async fn dead_one(&self, id: Uuid) -> Result<DeadMessage> {
107        let dialect = self.db.driver().dialect();
108        let sql = format!(
109            "SELECT {COLUMNS} FROM {} WHERE id = {}",
110            dialect.qualified("sofie", "messages"),
111            dialect.placeholder(1),
112        );
113        let params = [SqlValue::Uuid(id)];
114        let found = query_decoded(self.db.driver(), &sql, &params, Vec::new(), Stored::decode)
115            .await?
116            .pop()
117            .ok_or_else(|| Error::Rule(format!("there is no message {id}")))?;
118        let name = found.name.clone();
119        found.dead().ok_or_else(|| {
120            Error::Rule(format!(
121                "message {id} ({name}) is not dead: it waits or runs, and only a dead one is \
122                 replayed"
123            ))
124        })
125    }
126}
sofie_db/dead_messages.rs

sofie_db/
dead_messages.rs
