sofie_db/
product_events.rs

1use std::sync::Arc;
2use std::sync::atomic::{AtomicU64, Ordering};
3
4use sofie_core::{Date, Error, Result, SqlValue, Timestamp};
5use sofie_telemetry::{ProductRecord, keep_product_events};
6use tokio::sync::{mpsc, oneshot};
7
8use crate::{Db, Dialect, query_values};
9
10/// Past it, events are dropped rather than slowing the code that raises them.
11const QUEUED: usize = 8192;
12const BATCH: usize = 512;
13
14/// `cohorts` answers `start, offset, count` rows, an offset of -1 counting the cohort.
15pub struct ProductSql {
16    pub insert: &'static str,
17    pub cohorts: &'static str,
18    /// Deletes at most a batch, answering a row per event deleted.
19    pub prune: &'static str,
20    pub forget: &'static str,
21    /// Deletes what a write racing `forget` kept.
22    pub sweep: &'static str,
23}
24
25#[derive(Clone, Copy, Debug, PartialEq, Eq)]
26pub enum Period {
27    Day,
28    Week,
29    Month,
30}
31
32impl Period {
33    #[must_use]
34    pub fn parse(text: &str) -> Option<Self> {
35        match text {
36            "day" => Some(Self::Day),
37            "week" => Some(Self::Week),
38            "month" => Some(Self::Month),
39            _ => None,
40        }
41    }
42
43    #[must_use]
44    pub fn name(self) -> &'static str {
45        match self {
46            Self::Day => "day",
47            Self::Week => "week",
48            Self::Month => "month",
49        }
50    }
51}
52
53/// Principals by the period of their first `start` since `since`, then who raised `returning`.
54pub struct Retention<'a> {
55    pub start: &'a str,
56    pub returning: &'a str,
57    pub period: Period,
58    pub since: Timestamp,
59}
60
61#[derive(Clone, Debug, PartialEq, Eq)]
62pub struct Cohort {
63    pub start: Date,
64    pub size: u64,
65    pub returned: Vec<u64>,
66}
67
68/// Only events raised by a signed-in principal join a cohort.
69pub async fn cohorts(db: &Db, retention: &Retention<'_>) -> Result<Vec<Cohort>> {
70    let sql = product_sql(db.driver().dialect())?;
71    let params = [
72        SqlValue::Text(retention.period.name().to_owned()),
73        SqlValue::Text(retention.start.to_owned()),
74        SqlValue::Text(retention.returning.to_owned()),
75        SqlValue::Timestamp(retention.since),
76    ];
77    let rows = query_values(db.driver(), sql.cohorts, &params).await?;
78    let mut cohorts: Vec<Cohort> = Vec::new();
79    for row in rows {
80        let [
81            SqlValue::DateTime(start),
82            SqlValue::Int(offset),
83            SqlValue::Int(count),
84        ] = row.as_slice()
85        else {
86            return Err(Error::internal_message(format!(
87                "a cohort row has unexpected columns: {:?}",
88                row.iter().map(SqlValue::kind).collect::<Vec<_>>()
89            )));
90        };
91        let count = u64::try_from(*count).unwrap_or_default();
92        let start = start.date();
93        if cohorts.last().is_none_or(|cohort| cohort.start != start) {
94            cohorts.push(Cohort {
95                start,
96                size: 0,
97                returned: Vec::new(),
98            });
99        }
100        let cohort = cohorts.last_mut().expect("a cohort was just pushed");
101        match usize::try_from(*offset) {
102            Err(_) => cohort.size = count,
103            Ok(offset) => {
104                if cohort.returned.len() <= offset {
105                    cohort.returned.resize(offset + 1, 0);
106                }
107                cohort.returned[offset] = count;
108            }
109        }
110    }
111    Ok(cohorts)
112}
113
114impl crate::Prune {
115    pub async fn product_events(&self, before: sofie_core::Instant) -> Result<u64> {
116        let sql = product_sql(self.db().driver().dialect())?;
117        query_values(self.db().driver(), sql.sweep, &[]).await?;
118        let params = [SqlValue::Timestamp(before.timestamp())];
119        let mut deleted = 0;
120        loop {
121            let batch = query_values(self.db().driver(), sql.prune, &params)
122                .await?
123                .len();
124            if batch == 0 {
125                return Ok(deleted);
126            }
127            deleted += u64::try_from(batch)
128                .map_err(|error| Error::internal("counting deleted events", error))?;
129        }
130    }
131}
132
133pub(crate) fn product_sql(dialect: &dyn Dialect) -> Result<&'static ProductSql> {
134    dialect
135        .product_events()
136        .ok_or_else(|| Error::internal_message("this database keeps no product events"))
137}
138
139enum Kept {
140    Event(ProductRecord),
141    Flush(oneshot::Sender<()>),
142}
143
144pub struct ProductEvents {
145    queue: mpsc::Sender<Kept>,
146}
147
148impl ProductEvents {
149    /// Once per process; it must run inside the async runtime.
150    pub fn keep(db: &Db) -> Result<Self> {
151        let sql = product_sql(db.driver().dialect())?;
152        let (queue, mut received) = mpsc::channel(QUEUED);
153        let raised = queue.clone();
154        let dropped = Arc::new(AtomicU64::new(0));
155        let installed = keep_product_events(move |record| {
156            if raised.try_send(Kept::Event(record)).is_ok() {
157                return;
158            }
159            let lost = dropped.fetch_add(1, Ordering::Relaxed) + 1;
160            if lost.is_power_of_two() {
161                tracing::warn!(
162                    product.dropped = lost,
163                    "product events dropped: the database keeps up with fewer"
164                );
165            }
166        });
167        if !installed {
168            return Err(Error::internal_message(
169                "product events are kept already in this process",
170            ));
171        }
172        let db = db.clone();
173        tokio::spawn(async move {
174            let mut batch = Vec::with_capacity(BATCH);
175            while let Some(first) = received.recv().await {
176                let mut flushed = Vec::new();
177                let mut next = Some(first);
178                while let Some(kept) = next {
179                    match kept {
180                        Kept::Event(record) => batch.push(record),
181                        Kept::Flush(done) => flushed.push(done),
182                    }
183                    next = (batch.len() < BATCH)
184                        .then(|| received.try_recv().ok())
185                        .flatten();
186                }
187                write(&db, sql, &mut batch).await;
188                for done in flushed {
189                    let _ = done.send(());
190                }
191            }
192        });
193        Ok(Self { queue })
194    }
195
196    pub async fn flush(&self) {
197        let (done, written) = oneshot::channel();
198        if self.queue.send(Kept::Flush(done)).await.is_ok() {
199            let _ = written.await;
200        }
201    }
202}
203
204async fn write(db: &Db, sql: &ProductSql, batch: &mut Vec<ProductRecord>) {
205    if batch.is_empty() {
206        return;
207    }
208    let mut at = Vec::with_capacity(batch.len());
209    let mut names = Vec::with_capacity(batch.len());
210    let mut principals = Vec::with_capacity(batch.len());
211    for record in batch.drain(..) {
212        at.push(SqlValue::Timestamp(record.at));
213        names.push(SqlValue::Text(record.name));
214        principals.push(record.principal.map_or(SqlValue::Null, SqlValue::Uuid));
215    }
216    let count = at.len();
217    let params = [
218        SqlValue::Array(at),
219        SqlValue::Array(names),
220        SqlValue::Array(principals),
221    ];
222    if let Err(error) = query_values(db.driver(), sql.insert, &params).await {
223        tracing::warn!(
224            error = %error,
225            product.lost = count,
226            "writing product events failed"
227        );
228    }
229}
sofie_db/product_events.rs

sofie_db/
product_events.rs
