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::{Actor, 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    pub actor: Option<Actor>,
40}
41
42impl FeedEntry {
43    fn decode(row: &mut dyn ColumnReader) -> Result<Self> {
44        Ok(Self {
45            position: row.int()?,
46            id: row.uuid()?,
47            name: row.text()?.to_owned(),
48            payload: row.json()?,
49            traceparent: Option::<String>::decode(row)?,
50            actor: Actor::decode(row)?,
51        })
52    }
53}
54
55pub trait FeedSource: Send + Sync {
56    fn producer(&self) -> &str;
57
58    fn after(&self, position: i64, limit: i64) -> BoxFuture<'_, Result<Vec<FeedEntry>>>;
59}
60
61pub(crate) fn feed_sql(dialect: &dyn Dialect) -> Result<&'static FeedSql> {
62    dialect.feed().ok_or_else(|| {
63        Error::internal_message(
64            "the event feed of split apps needs Postgres: this database keeps none, so its \
65             worker may neither publish nor pull events",
66        )
67    })
68}
69
70pub fn keeps_feed(db: &Db) -> Result<()> {
71    feed_sql(db.driver().dialect()).map(|_| ())
72}
73
74pub async fn read_feed(db: &Db, after: i64, limit: i64) -> Result<Vec<FeedEntry>> {
75    let params = [
76        SqlValue::Int(after),
77        SqlValue::Int(limit.clamp(1, FEED_PAGE)),
78    ];
79    query_decoded(
80        db.driver(),
81        feed_sql(db.driver().dialect())?.read,
82        &params,
83        Vec::new(),
84        FeedEntry::decode,
85    )
86    .await
87}
88
89pub async fn feed_cursor(db: &Db, producer: &str) -> Result<i64> {
90    let params = [SqlValue::Text(producer.to_owned())];
91    let sql = feed_sql(db.driver().dialect())?.cursor;
92    let found = query_decoded(db.driver(), sql, &params, Vec::new(), |row| row.int());
93    Ok(found.await?.into_iter().next().unwrap_or(0))
94}
95
96/// A consumer racing from the same cursor fails as `Conflict`. Before any worker is published,
97/// entries wait as events; one no listener hears is dropped with a warning after an hour.
98pub async fn receive_feed(db: &Db, producer: &str, from: i64, entries: &[FeedEntry]) -> Result<()> {
99    let Some(last) = entries.last() else {
100        return Ok(());
101    };
102    let mut received = Received {
103        producer,
104        from,
105        to: last.position,
106        entries,
107    };
108    db.commit(&mut received).await
109}
110
111struct Received<'a> {
112    producer: &'a str,
113    from: i64,
114    to: i64,
115    entries: &'a [FeedEntry],
116}
117
118impl Received<'_> {
119    /// Keyed by the producer's id, so an entry received again adds no job.
120    fn fan_out(&self, plan: &mut Plan<'_>, listeners: &Listeners) -> Result<()> {
121        let entries = self.entries.iter().map(|entry| Fanned {
122            id: entry.id,
123            name: &entry.name,
124            payload: entry.payload.clone(),
125            traceparent: entry.traceparent.clone(),
126            actor: entry.actor.as_ref(),
127        });
128        let now = plan.stamp().now.timestamp();
129        if let Some(step) = fan_out_events(plan.dialect(), entries, now, listeners)? {
130            plan.push_message(step);
131        }
132        Ok(())
133    }
134
135    fn queue(&self, plan: &mut Plan<'_>) -> Result<()> {
136        let feed = feed_sql(plan.dialect())?;
137        let now = SqlValue::Timestamp(plan.stamp().now.timestamp());
138        for entry in self.entries {
139            let mut params = vec![
140                SqlValue::Uuid(entry.id),
141                SqlValue::Text(entry.name.clone()),
142                SqlValue::Json(entry.payload.clone()),
143                now.clone(),
144                entry
145                    .traceparent
146                    .clone()
147                    .map_or(SqlValue::Null, SqlValue::Text),
148            ];
149            params.extend(Actor::columns(entry.actor.as_ref()));
150            plan.push_message(Step::statement(feed.receive, params, None));
151        }
152        Ok(())
153    }
154}
155
156impl Outcome for Received<'_> {
157    fn plan(&self, plan: &mut Plan<'_>) -> Result<()> {
158        let feed = feed_sql(plan.dialect())?;
159        match plan.heard.clone() {
160            Some(heard) => self.fan_out(plan, &heard.listeners)?,
161            None => self.queue(plan)?,
162        }
163        let cursor = vec![
164            SqlValue::Text(self.producer.to_owned()),
165            SqlValue::Int(self.from),
166            SqlValue::Int(self.to),
167        ];
168        plan.push(Step::statement(feed.advance, cursor, Some(1)));
169        #[cfg(feature = "live")]
170        plan.announcements.live_change(crate::TableChange {
171            topic: crate::Topic::remote(self.producer),
172            readers: crate::Readers::Anyone,
173        });
174        Ok(())
175    }
176}
177
178/// Whether a full page came, so another may follow at once.
179pub(crate) async fn pull_once(db: &Db, source: &dyn FeedSource) -> Result<bool> {
180    let from = feed_cursor(db, source.producer()).await?;
181    let entries = source.after(from, FEED_PAGE).await?;
182    receive_feed(db, source.producer(), from, &entries).await?;
183    Ok(i64::try_from(entries.len()).is_ok_and(|count| count >= FEED_PAGE))
184}
185
186pub(crate) async fn pull(db: Db, source: Arc<dyn FeedSource>, mut stopped: watch::Receiver<bool>) {
187    loop {
188        let wait = match pull_once(&db, source.as_ref()).await {
189            Ok(true) => Duration::ZERO,
190            Ok(false) => PULL_INTERVAL,
191            Err(error) => {
192                tracing::warn!(
193                    error.kind = "feed",
194                    feed.producer = source.producer(),
195                    error = %error,
196                    "pulling another app's events failed; trying again"
197                );
198                PULL_INTERVAL_AFTER_FAILURE
199            }
200        };
201        if *stopped.borrow() {
202            return;
203        }
204        if tokio::time::timeout(wait, stopped.changed()).await.is_ok() {
205            return;
206        }
207    }
208}
209
210struct Appended {
211    id: Uuid,
212    name: &'static str,
213    payload: Value,
214    actor: Option<Actor>,
215}
216
217impl Outcome for Appended {
218    fn plan(&self, plan: &mut Plan<'_>) -> Result<()> {
219        let mut params = vec![
220            SqlValue::Uuid(self.id),
221            SqlValue::Text(self.name.to_owned()),
222            SqlValue::Json(self.payload.clone()),
223            current_traceparent(),
224        ];
225        params.extend(Actor::columns(self.actor.as_ref()));
226        plan.push(Step::statement(
227            feed_sql(plan.dialect())?.append,
228            params,
229            None,
230        ));
231        Ok(())
232    }
233}
234
235/// Appended with the deletion of the message that carried it, so it is appended once.
236pub(crate) fn append(event: &'static str) -> Run<JobContext> {
237    Arc::new(move |context: &mut JobContext| {
238        let appended = Appended {
239            id: context.message,
240            name: event,
241            payload: context.payload.clone(),
242            actor: context.actor.clone(),
243        };
244        Box::pin(std::future::ready(Ok(
245            Box::new(appended) as Box<dyn AnyOutcome>
246        )))
247    })
248}
249
250struct Pruned;
251
252impl Outcome for Pruned {
253    fn plan(&self, plan: &mut Plan<'_>) -> Result<()> {
254        let sql = feed_sql(plan.dialect())?;
255        let before = plan
256            .stamp()
257            .now
258            .timestamp()
259            .checked_sub(FEED_RETENTION)
260            .map_err(|error| Error::internal("computing the feed's retention", error))?;
261        plan.push(Step::statement(
262            sql.prune,
263            vec![SqlValue::Timestamp(before)],
264            None,
265        ));
266        Ok(())
267    }
268}
269
270pub(crate) fn prune() -> Run<JobContext> {
271    crate::erase(|| async { Ok(Pruned) })
272}
sofie_db/feed.rs

sofie_db/
feed.rs
