sofie_db/
messages.rs

1use std::time::Duration;
2
3use serde_json::{Value, json};
4use sofie_core::variants::{self, Variants};
5use sofie_core::{Cast, ColumnReader, Error, Result, SqlValue, Timestamp, Uuid};
6use sofie_telemetry::TraceParent;
7
8use crate::heard::{Known, Listeners};
9use crate::{BoxFuture, Db, Dialect, Stamp, Step, Wakeups, query_decoded, query_values};
10
11/// Every insert binds the message columns first, as `$1` to `$8`: id, kind,
12/// name, payload, `available_at`, traceparent, actor, actor's area.
13pub struct MessageSql {
14    pub insert: &'static str,
15    /// Message columns, then a key; skipped while a job of that name waits for it.
16    pub fan_out: &'static str,
17    /// `(events, jobs, available_at)`: events `{id, payload, traceparent, actor, actor_area}`,
18    /// jobs `{id, name, event}`.
19    pub fan_out_events: &'static str,
20    /// `(lease_milliseconds, now, limit, mine, held_until)`, counting attempts past the last;
21    /// `mine` names by kind, and any other waits until `held_until`.
22    pub claim: &'static str,
23    /// `(id, attempts, lease_milliseconds)`: extends a claim still held, on the database's clock.
24    pub renew: &'static str,
25    pub complete: &'static str,
26    pub retry: &'static str,
27    /// Clears the key, so a dead job holds back no new one for it.
28    pub bury: &'static str,
29    /// `(id, attempts, name)`: turns an event its lone listener ran into that listener's job.
30    pub to_job: &'static str,
31    pub add_schedule: &'static str,
32    /// Message columns, then the next time; returns it and whether this call queued a run (none
33    /// while one waits or retries), no row when the schedule does not exist.
34    pub fire_schedule: &'static str,
35    /// Sent after a commit that queued messages, never within it: Postgres
36    /// serializes the commits of notifying transactions on one lock.
37    pub notify: Option<&'static str>,
38}
39
40#[derive(Clone, Copy, Debug, PartialEq, Eq)]
41pub enum MessageKind {
42    Event,
43    Job,
44}
45
46impl Variants for MessageKind {
47    const TYPE: &'static str = "message kind";
48    const VARIANTS: &'static [(&'static str, Self)] = &[("event", Self::Event), ("job", Self::Job)];
49}
50
51/// Its `attempt` fences every later write, so an expired lease can neither finish nor fail it.
52#[derive(Debug)]
53pub struct Claimed {
54    pub id: Uuid,
55    pub kind: MessageKind,
56    pub name: String,
57    pub payload: Value,
58    pub attempt: i64,
59    pub traceparent: Option<TraceParent>,
60    pub available_at: Timestamp,
61    pub actor: Option<Actor>,
62}
63
64#[derive(Clone, Debug, PartialEq, Eq)]
65#[cfg_attr(
66    feature = "remote-contracts",
67    derive(serde::Serialize, serde::Deserialize)
68)]
69pub struct Actor {
70    pub principal: Uuid,
71    pub area: Option<String>,
72}
73
74impl Actor {
75    pub(crate) fn queuing(principal: Option<Uuid>) -> Option<Self> {
76        principal.map(|principal| Self {
77            principal,
78            area: acting_area(principal),
79        })
80    }
81
82    pub(crate) fn decode(row: &mut dyn ColumnReader) -> Result<Option<Self>> {
83        let principal = Option::<Uuid>::decode(row)?;
84        let area = Option::<String>::decode(row)?;
85        Ok(principal.map(|principal| Self { principal, area }))
86    }
87
88    pub(crate) fn columns(actor: Option<&Self>) -> [SqlValue; 2] {
89        let Some(actor) = actor else {
90            return [SqlValue::Null, SqlValue::Null];
91        };
92        let area = actor.area.clone().map_or(SqlValue::Null, SqlValue::Text);
93        [SqlValue::Uuid(actor.principal), area]
94    }
95
96    #[cfg(feature = "remote-contracts")]
97    pub(crate) fn acting(&self) -> Option<sofie_auth::Acting> {
98        self.area.clone().map(|area| sofie_auth::Acting {
99            area,
100            principal: self.principal,
101        })
102    }
103}
104
105#[cfg(feature = "remote-contracts")]
106fn acting_area(principal: Uuid) -> Option<String> {
107    crate::remote::acting()
108        .filter(|acting| acting.principal == principal)
109        .map(|acting| acting.area)
110}
111
112#[cfg(not(feature = "remote-contracts"))]
113fn acting_area(_principal: Uuid) -> Option<String> {
114    None
115}
116
117impl Claimed {
118    fn decode(row: &mut dyn ColumnReader) -> Result<Self> {
119        Ok(Self {
120            id: row.uuid()?,
121            kind: variants::decode(row)?,
122            name: row.text()?.to_owned(),
123            payload: row.json()?,
124            attempt: row.int()?,
125            traceparent: traceparent(row)?,
126            available_at: row.timestamp()?,
127            actor: Actor::decode(row)?,
128        })
129    }
130
131    fn fenced(&self, more: impl IntoIterator<Item = SqlValue>) -> Vec<SqlValue> {
132        let mut params = vec![SqlValue::Uuid(self.id), SqlValue::Int(self.attempt)];
133        params.extend(more);
134        params
135    }
136}
137
138fn traceparent(row: &mut dyn ColumnReader) -> Result<Option<TraceParent>> {
139    if row.is_null()? {
140        row.skip()?;
141        return Ok(None);
142    }
143    Ok(TraceParent::parse(row.text()?))
144}
145
146#[derive(Clone, Copy, Debug, PartialEq, Eq)]
147pub struct Fired {
148    pub fired: bool,
149    pub next_at: Timestamp,
150}
151
152impl Fired {
153    fn decode(row: &mut dyn ColumnReader) -> Result<Self> {
154        Ok(Self {
155            next_at: row.timestamp()?,
156            fired: row.bool()?,
157        })
158    }
159}
160
161struct Message<'a> {
162    kind: MessageKind,
163    name: &'a str,
164    payload: Value,
165    available_at: Timestamp,
166    actor: Option<&'a Actor>,
167}
168
169impl Message<'_> {
170    fn columns(self) -> Result<Vec<SqlValue>> {
171        let mut columns = vec![
172            SqlValue::Uuid(Uuid::now_v7()?),
173            SqlValue::Text(self.kind.name().to_owned()),
174            SqlValue::Text(self.name.to_owned()),
175            SqlValue::Json(self.payload),
176            SqlValue::Timestamp(self.available_at),
177            current_traceparent(),
178        ];
179        columns.extend(Actor::columns(self.actor));
180        Ok(columns)
181    }
182}
183
184pub(crate) fn keyed_job(
185    dialect: &dyn Dialect,
186    name: &str,
187    payload: Value,
188    key: Uuid,
189    available_at: Timestamp,
190    principal: Option<Uuid>,
191) -> Result<Step> {
192    let actor = Actor::queuing(principal);
193    let mut params = Message {
194        kind: MessageKind::Job,
195        name,
196        payload,
197        available_at,
198        actor: actor.as_ref(),
199    }
200    .columns()?;
201    params.push(SqlValue::Uuid(key));
202    Ok(Step::statement(dialect.messages().fan_out, params, None))
203}
204
205/// Due times follow the app's clock, so a test can move them; leases follow the database's.
206#[derive(Clone)]
207pub struct Messages {
208    db: Db,
209}
210
211impl Messages {
212    #[must_use]
213    pub fn new(db: Db) -> Self {
214        Self { db }
215    }
216
217    fn sql(&self) -> &'static MessageSql {
218        self.db.driver().dialect().messages()
219    }
220
221    pub async fn claim(&self, lease: Duration, limit: i64) -> Result<Vec<Claimed>> {
222        let now = self.db.clock().now().timestamp();
223        let nothing_known = SqlValue::Json(Known::default().to_json());
224        claim_known(&self.db, lease, limit, &nothing_known, now).await
225    }
226
227    /// A conflict means the claim was lost: another worker took the message after its lease ran out.
228    pub async fn renew(&self, message: &Claimed, lease: Duration) -> Result<()> {
229        let params = message.fenced([milliseconds(lease)?]);
230        let step = Step::statement(self.sql().renew, params, Some(1));
231        self.db.transaction(vec![step]).await
232    }
233
234    /// Committed with its job's outcome, it makes the outcome happen once.
235    #[must_use]
236    pub fn completion(&self, message: &Claimed) -> Step {
237        Step::statement(self.sql().complete, message.fenced([]), Some(1))
238    }
239
240    /// `listener` names the lone listener an event ran as; the message becomes its job.
241    pub async fn retry(
242        &self,
243        message: &Claimed,
244        listener: Option<&str>,
245        error: &str,
246        delay: Duration,
247    ) -> Result<()> {
248        let due = later(self.db.clock().now().timestamp(), delay)?;
249        let due = Some(SqlValue::Timestamp(due));
250        self.release(self.sql().retry, message, listener, error, due)
251            .await
252    }
253
254    pub async fn bury(&self, message: &Claimed, listener: Option<&str>, error: &str) -> Result<()> {
255        self.release(self.sql().bury, message, listener, error, None)
256            .await
257    }
258
259    /// A lone listener's message first becomes its job, so its retries and its death name the
260    /// listener that failed, not the event.
261    async fn release(
262        &self,
263        sql: &'static str,
264        message: &Claimed,
265        listener: Option<&str>,
266        error: &str,
267        due: Option<SqlValue>,
268    ) -> Result<()> {
269        let params = message.fenced(std::iter::once(SqlValue::Text(error.to_owned())).chain(due));
270        let mut steps = Vec::with_capacity(2);
271        if let Some(listener) = listener {
272            let params = message.fenced([SqlValue::Text(listener.to_owned())]);
273            steps.push(Step::statement(self.sql().to_job, params, Some(1)));
274        }
275        steps.push(Step::statement(sql, params, Some(1)));
276        self.db.transaction(steps).await
277    }
278
279    pub async fn add_schedule(&self, name: &str, first_at: Timestamp, cadence: &str) -> Result<()> {
280        let params = [
281            SqlValue::Text(name.to_owned()),
282            SqlValue::Timestamp(first_at),
283            SqlValue::Text(cadence.to_owned()),
284        ];
285        let sql = self.sql().add_schedule;
286        query_values(self.db.driver(), sql, &params).await.map(drop)
287    }
288
289    /// A run by hand: a running worker keeps the next due time in memory, but takes a job at once.
290    pub async fn queue_schedule(&self, name: &str) -> Result<()> {
291        let dialect = self.db.driver().dialect();
292        let payload = Value::Object(serde_json::Map::new());
293        let now = self.db.clock().now().timestamp();
294        let mut steps = Vec::new();
295        let queuing = Queuing {
296            kind: MessageKind::Job,
297            at: now,
298            actor: None,
299        };
300        queue_steps(dialect, &queuing, [(name, payload)], None, &mut steps)?;
301        // In the same transaction, as the process may end before a notification sent later.
302        if let Some(notify) = dialect.messages().notify {
303            steps.push(Step::Script(notify.to_owned()));
304        }
305        self.db.transaction(steps).await
306    }
307
308    /// Only one of any number of concurrent callers is told it fired.
309    pub async fn fire_schedule(
310        &self,
311        name: &str,
312        now: Timestamp,
313        next_at: Timestamp,
314    ) -> Result<Option<Fired>> {
315        let payload = Value::Object(serde_json::Map::new());
316        let mut params = Message {
317            kind: MessageKind::Job,
318            name,
319            payload,
320            available_at: now,
321            actor: None,
322        }
323        .columns()?;
324        params.push(SqlValue::Timestamp(next_at));
325        let sql = self.sql().fire_schedule;
326        let fired =
327            query_decoded(self.db.driver(), sql, &params, Vec::new(), Fired::decode).await?;
328        Ok(fired.into_iter().next())
329    }
330
331    pub(crate) async fn arrived(&self) {
332        self.db.arrivals().arrived().await;
333    }
334
335    #[must_use]
336    pub fn wakeups(&self) -> Option<BoxFuture<'_, Result<Box<dyn Wakeups>>>> {
337        self.db.driver().wakeups()
338    }
339}
340
341pub(crate) async fn claim_known(
342    db: &Db,
343    lease: Duration,
344    limit: i64,
345    mine: &SqlValue,
346    held_until: Timestamp,
347) -> Result<Vec<Claimed>> {
348    let params = [
349        milliseconds(lease)?,
350        SqlValue::Timestamp(db.clock().now().timestamp()),
351        SqlValue::Int(limit),
352        mine.clone(),
353        SqlValue::Timestamp(held_until),
354    ];
355    let sql = db.driver().dialect().messages().claim;
356    query_decoded(db.driver(), sql, &params, Vec::new(), Claimed::decode).await
357}
358
359pub async fn ping_queue(db: &Db) -> Result<()> {
360    let messages = db.driver().dialect().qualified("sofie", "messages");
361    let sql = format!("SELECT 1 FROM {messages} LIMIT 1");
362    query_values(db.driver(), &sql, &[]).await.map(drop)
363}
364
365pub(crate) struct Queuing {
366    kind: MessageKind,
367    at: Timestamp,
368    actor: Option<Actor>,
369}
370
371impl Stamp {
372    pub(crate) fn queuing(&self, kind: MessageKind, at: Timestamp) -> Queuing {
373        Queuing {
374            kind,
375            at,
376            actor: Actor::queuing(self.actor),
377        }
378    }
379}
380
381pub(crate) fn queue_steps<'a>(
382    dialect: &dyn Dialect,
383    queuing: &Queuing,
384    messages: impl IntoIterator<Item = (&'a str, Value)>,
385    heard: Option<&Known>,
386    steps: &mut Vec<Step>,
387) -> Result<bool> {
388    if let Some(heard) = heard.filter(|_| queuing.kind == MessageKind::Event) {
389        return fan_out_raised(dialect, queuing, messages, &heard.listeners, steps);
390    }
391    let mut queued = false;
392    for (name, payload) in messages {
393        let params = Message {
394            kind: queuing.kind,
395            name,
396            payload,
397            available_at: queuing.at,
398            actor: queuing.actor.as_ref(),
399        }
400        .columns()?;
401        steps.push(Step::statement(dialect.messages().insert, params, None));
402        queued = true;
403    }
404    Ok(queued)
405}
406
407fn fan_out_raised<'a>(
408    dialect: &dyn Dialect,
409    queuing: &Queuing,
410    events: impl IntoIterator<Item = (&'a str, Value)>,
411    listeners: &Listeners,
412    steps: &mut Vec<Step>,
413) -> Result<bool> {
414    let traceparent = sofie_telemetry::current_traceparent().map(|parent| parent.to_string());
415    let mut raised = Vec::new();
416    for (name, payload) in events {
417        raised.push(Fanned {
418            id: Uuid::now_v7()?,
419            name,
420            payload,
421            traceparent: traceparent.clone(),
422            actor: queuing.actor.as_ref(),
423        });
424    }
425    let Some(step) = fan_out_events(dialect, raised, queuing.at, listeners)? else {
426        return Ok(false);
427    };
428    steps.push(step);
429    Ok(true)
430}
431
432pub(crate) struct Fanned<'a> {
433    pub(crate) id: Uuid,
434    pub(crate) name: &'a str,
435    pub(crate) payload: Value,
436    pub(crate) traceparent: Option<String>,
437    pub(crate) actor: Option<&'a Actor>,
438}
439
440pub(crate) fn fan_out_events<'a>(
441    dialect: &dyn Dialect,
442    events: impl IntoIterator<Item = Fanned<'a>>,
443    at: Timestamp,
444    listeners: &Listeners,
445) -> Result<Option<Step>> {
446    let (mut taken, mut jobs) = (Vec::new(), Vec::new());
447    for event in events {
448        let taking = listeners.taking(event.name, &event.payload);
449        if taking.is_empty() {
450            continue;
451        }
452        let id = event.id.to_string();
453        for listener in taking.iter() {
454            let job = Uuid::now_v7()?.to_string();
455            jobs.push(json!({ "id": job, "name": listener, "event": id }));
456        }
457        taken.push(json!({
458            "id": id,
459            "payload": event.payload,
460            "traceparent": event.traceparent,
461            "actor": event.actor.map(|actor| actor.principal.to_string()),
462            "actor_area": event.actor.and_then(|actor| actor.area.as_deref()),
463        }));
464    }
465    if jobs.is_empty() {
466        return Ok(None);
467    }
468    let params = vec![
469        SqlValue::Json(Value::Array(taken)),
470        SqlValue::Json(Value::Array(jobs)),
471        SqlValue::Timestamp(at),
472    ];
473    let sql = dialect.messages().fan_out_events;
474    Ok(Some(Step::statement(sql, params, None)))
475}
476
477pub(crate) fn current_traceparent() -> SqlValue {
478    sofie_telemetry::current_traceparent().map_or(SqlValue::Null, |traceparent| {
479        SqlValue::Text(traceparent.to_string())
480    })
481}
482
483fn milliseconds(lease: Duration) -> Result<SqlValue> {
484    i64::try_from(lease.as_millis())
485        .map(SqlValue::Int)
486        .map_err(|error| Error::internal("a lease too long to store", error))
487}
488
489fn later(now: Timestamp, by: Duration) -> Result<Timestamp> {
490    now.checked_add(by)
491        .map_err(|error| Error::internal("computing a message time", error))
492}
sofie_db/messages.rs

sofie_db/
messages.rs
