sofie_db/
feed.rs

1use std::sync::Arc;
2use std::time::Duration;
3
4use serde::{Deserialize, Serialize};
5use serde_json::Value;
6use sofie_core::{Cast, ColumnReader, Error, Result, SqlValue, Uuid};
7use tokio::sync::watch;
8
9use crate::heard::Listeners;
10use crate::messages::{Fanned, current_traceparent, fan_out_events};
11use crate::worker::JobContext;
12use crate::{AnyOutcome, BoxFuture, Db, Dialect, Outcome, Plan, Run, Step, query_decoded};
13
14/// Entries older than this are pruned; a consumer further behind misses them.
15const FEED_RETENTION: Duration = Duration::from_hours(7 * 24);
16pub const FEED_PAGE: i64 = 100;
17const PULL_INTERVAL: Duration = Duration::from_secs(1);
18const PULL_INTERVAL_AFTER_FAILURE: Duration = Duration::from_secs(5);
19
20/// `append` is serialized by a lock, so a reader past a position never misses an earlier one;
21/// `receive` skips an id already queued; `advance` moves a cursor only from where it was.
22pub struct FeedSql {
23    pub append: &'static str,
24    pub read: &'static str,
25    pub prune: &'static str,
26    pub receive: &'static str,
27    pub cursor: &'static str,
28    pub advance: &'static str,
29}
30
31/// An event as it crosses apps: its id stays the producer's, so it is delivered once.
32#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
33pub struct FeedEntry {
34    pub position: i64,
35    pub id: Uuid,
36    pub name: String,
37    pub payload: Value,
38    pub traceparent: Option<String>,
39}
40
41impl FeedEntry {
42    fn decode(row: &mut dyn ColumnReader) -> Result<Self> {
43        Ok(Self {
44            position: row.int()?,
45            id: row.uuid()?,
46            name: row.text()?.to_owned(),
47            payload: row.json()?,
48            traceparent: Option::<String>::decode(row)?,
49        })
50    }
51}
52
53pub trait FeedSource: Send + Sync {
54    fn producer(&self) -> &str;
55
56    fn after(&self, position: i64, limit: i64) -> BoxFuture<'_, Result<Vec<FeedEntry>>>;
57}
58
59fn feed_sql(dialect: &dyn Dialect) -> Result<&'static FeedSql> {
60    dialect
61        .feed()
62        .ok_or_else(|| Error::internal_message("this database keeps no event feed"))
63}
64
65pub async fn read_feed(db: &Db, after: i64, limit: i64) -> Result<Vec<FeedEntry>> {
66    let params = [
67        SqlValue::Int(after),
68        SqlValue::Int(limit.clamp(1, FEED_PAGE)),
69    ];
70    query_decoded(
71        db.driver(),
72        feed_sql(db.driver().dialect())?.read,
73        &params,
74        Vec::new(),
75        FeedEntry::decode,
76    )
77    .await
78}
79
80pub async fn feed_cursor(db: &Db, producer: &str) -> Result<i64> {
81    let params = [SqlValue::Text(producer.to_owned())];
82    let sql = feed_sql(db.driver().dialect())?.cursor;
83    let found = query_decoded(db.driver(), sql, &params, Vec::new(), |row| row.int());
84    Ok(found.await?.into_iter().next().unwrap_or(0))
85}
86
87/// A consumer racing from the same cursor fails as `Conflict`. Before any worker is published,
88/// entries wait as events; one no listener hears is dropped with a warning after an hour.
89pub async fn receive_feed(db: &Db, producer: &str, from: i64, entries: &[FeedEntry]) -> Result<()> {
90    let Some(last) = entries.last() else {
91        return Ok(());
92    };
93    let mut received = Received {
94        producer,
95        from,
96        to: last.position,
97        entries,
98    };
99    db.commit(&mut received).await
100}
101
102struct Received<'a> {
103    producer: &'a str,
104    from: i64,
105    to: i64,
106    entries: &'a [FeedEntry],
107}
108
109impl Received<'_> {
110    /// Keyed by the producer's id, so an entry received again adds no job.
111    fn fan_out(&self, plan: &mut Plan<'_>, listeners: &Listeners) -> Result<()> {
112        let entries = self.entries.iter().map(|entry| Fanned {
113            id: entry.id,
114            name: &entry.name,
115            payload: entry.payload.clone(),
116            traceparent: entry.traceparent.clone(),
117        });
118        let now = plan.stamp().now.timestamp();
119        if let Some(step) = fan_out_events(plan.dialect(), entries, now, None, listeners)? {
120            plan.push_message(step);
121        }
122        Ok(())
123    }
124
125    fn queue(&self, plan: &mut Plan<'_>) -> Result<()> {
126        let feed = feed_sql(plan.dialect())?;
127        let now = SqlValue::Timestamp(plan.stamp().now.timestamp());
128        for entry in self.entries {
129            let params = vec![
130                SqlValue::Uuid(entry.id),
131                SqlValue::Text(entry.name.clone()),
132                SqlValue::Json(entry.payload.clone()),
133                now.clone(),
134                entry
135                    .traceparent
136                    .clone()
137                    .map_or(SqlValue::Null, SqlValue::Text),
138            ];
139            plan.push_message(Step::statement(feed.receive, params, None));
140        }
141        Ok(())
142    }
143}
144
145impl Outcome for Received<'_> {
146    fn plan(&self, plan: &mut Plan<'_>) -> Result<()> {
147        let feed = feed_sql(plan.dialect())?;
148        match plan.heard.clone() {
149            Some(heard) => self.fan_out(plan, &heard.listeners)?,
150            None => self.queue(plan)?,
151        }
152        let cursor = vec![
153            SqlValue::Text(self.producer.to_owned()),
154            SqlValue::Int(self.from),
155            SqlValue::Int(self.to),
156        ];
157        plan.push(Step::statement(feed.advance, cursor, Some(1)));
158        Ok(())
159    }
160}
161
162/// Whether a full page came, so another may follow at once.
163pub(crate) async fn pull_once(db: &Db, source: &dyn FeedSource) -> Result<bool> {
164    let from = feed_cursor(db, source.producer()).await?;
165    let entries = source.after(from, FEED_PAGE).await?;
166    receive_feed(db, source.producer(), from, &entries).await?;
167    Ok(i64::try_from(entries.len()).is_ok_and(|count| count >= FEED_PAGE))
168}
169
170pub(crate) async fn pull(db: Db, source: Arc<dyn FeedSource>, mut stopped: watch::Receiver<bool>) {
171    loop {
172        let wait = match pull_once(&db, source.as_ref()).await {
173            Ok(true) => Duration::ZERO,
174            Ok(false) => PULL_INTERVAL,
175            Err(error) => {
176                tracing::warn!(
177                    error.kind = "feed",
178                    feed.producer = source.producer(),
179                    error = %error,
180                    "pulling another app's events failed; trying again"
181                );
182                PULL_INTERVAL_AFTER_FAILURE
183            }
184        };
185        if *stopped.borrow() {
186            return;
187        }
188        if tokio::time::timeout(wait, stopped.changed()).await.is_ok() {
189            return;
190        }
191    }
192}
193
194struct Appended {
195    id: Uuid,
196    name: &'static str,
197    payload: Value,
198}
199
200impl Outcome for Appended {
201    fn plan(&self, plan: &mut Plan<'_>) -> Result<()> {
202        let params = vec![
203            SqlValue::Uuid(self.id),
204            SqlValue::Text(self.name.to_owned()),
205            SqlValue::Json(self.payload.clone()),
206            current_traceparent(),
207        ];
208        plan.push(Step::statement(
209            feed_sql(plan.dialect())?.append,
210            params,
211            None,
212        ));
213        Ok(())
214    }
215}
216
217/// Appended with the deletion of the message that carried it, so it is appended once.
218pub(crate) fn append(event: &'static str) -> Run<JobContext> {
219    Arc::new(move |context: &mut JobContext| {
220        let appended = Appended {
221            id: context.message,
222            name: event,
223            payload: context.payload.clone(),
224        };
225        Box::pin(std::future::ready(Ok(
226            Box::new(appended) as Box<dyn AnyOutcome>
227        )))
228    })
229}
230
231struct Pruned;
232
233impl Outcome for Pruned {
234    fn plan(&self, plan: &mut Plan<'_>) -> Result<()> {
235        let sql = feed_sql(plan.dialect())?;
236        let before = plan
237            .stamp()
238            .now
239            .timestamp()
240            .checked_sub(FEED_RETENTION)
241            .map_err(|error| Error::internal("computing the feed's retention", error))?;
242        plan.push(Step::statement(
243            sql.prune,
244            vec![SqlValue::Timestamp(before)],
245            None,
246        ));
247        Ok(())
248    }
249}
250
251pub(crate) fn prune() -> Run<JobContext> {
252    crate::erase(|| async { Ok(Pruned) })
253}
sofie_db/feed.rs

sofie_db/
feed.rs
