sofie_db/
prune.rs

1use sofie_core::{Error, Ordered, Result, SqlValue};
2
3use crate::heard::Listeners;
4use crate::worker::JobContext;
5use crate::{Aggregate, Column, Context, Db, Dialect, Outcome, Plan, Resolve, Select, Step};
6
7const BATCH: u64 = 1000;
8
9/// Taken by a job, as `async fn forget_audits(prune: Prune, clock: Clock)`.
10pub struct Prune {
11    db: Db,
12}
13
14impl Resolve<JobContext> for Prune {
15    fn resolve(context: &mut JobContext) -> impl Future<Output = Result<Self>> + Send {
16        std::future::ready(Ok(Self {
17            db: context.db().clone(),
18        }))
19    }
20}
21
22impl Prune {
23    #[cfg(feature = "product-reports")]
24    pub(crate) fn db(&self) -> &Db {
25        &self.db
26    }
27
28    /// In batches, so no statement holds many rows' locks and a retried job resumes; nulls stay.
29    pub async fn before<A: Aggregate, T: Ordered + Clone + Send>(
30        &self,
31        column: Column<A, T>,
32        before: T,
33    ) -> Result<u64> {
34        #[cfg(feature = "children")]
35        if !A::CHILDREN.is_empty() {
36            return Err(Error::internal_message(format!(
37                "`{}` has children, which pruning would leave behind; delete its records with \
38                 their children in a job of the module's own",
39                A::NAME
40            )));
41        }
42        #[cfg(feature = "search-engine")]
43        let index = crate::search::index_of(&self.db, A::MODULE, A::NAME);
44        let heard = self.db.arrivals().heard().known().unwrap_or_default();
45        let mut deleted = 0;
46        loop {
47            let mut query = A::query()
48                .where_(column.lt(before.clone()))
49                .order_by(column.asc())
50                .limit(BATCH);
51            query.criteria.with_hidden = true;
52            let records = query.all(&self.db.read()).await?;
53            let batch = u64::try_from(records.len())
54                .map_err(|error| Error::internal("counting pruned records", error))?;
55            let events = match records.first() {
56                Some(first) => heard_events(first, &heard.listeners)?,
57                None => Events::default(),
58            };
59            self.db
60                .commit(&mut Pruned {
61                    records,
62                    events,
63                    #[cfg(feature = "search-engine")]
64                    index: index.clone(),
65                })
66                .await?;
67            deleted += batch;
68            if batch < BATCH {
69                return Ok(deleted);
70            }
71        }
72    }
73}
74
75#[cfg_attr(
76    not(feature = "contracts"),
77    expect(unused_variables, clippy::unnecessary_wraps)
78)]
79fn heard_events<A: Aggregate>(first: &A, heard: &Listeners) -> Result<Events> {
80    Ok(Events {
81        #[cfg(feature = "contracts")]
82        change: change_event(first, heard)?,
83        pruned: A::PRUNED.filter(|name| heard.hear(name)),
84    })
85}
86
87#[cfg(feature = "contracts")]
88fn change_event<A: Aggregate>(first: &A, heard: &Listeners) -> Result<Option<&'static str>> {
89    match A::CHANGED {
90        Some(name) if heard.hear_pruned(name) => {
91            Ok(crate::change_event::changes(first)?.then_some(name))
92        }
93        _ => Ok(None),
94    }
95}
96
97#[derive(Default)]
98struct Events {
99    #[cfg(feature = "contracts")]
100    change: Option<&'static str>,
101    pruned: Option<&'static str>,
102}
103
104struct Pruned<A> {
105    records: Vec<A>,
106    events: Events,
107    #[cfg(feature = "search-engine")]
108    index: Option<crate::SearchIndex>,
109}
110
111/// Each record only at the version it was loaded at: one changed since is a conflict, retried.
112impl<A: Aggregate> Outcome for Pruned<A> {
113    fn plan(&self, plan: &mut Plan<'_>) -> Result<()> {
114        if self.records.is_empty() {
115            return Ok(());
116        }
117        plan.push_as::<A>(delete_loaded::<A>(&self.records, plan.dialect())?);
118        let ids: Vec<sofie_core::Uuid> = self
119            .records
120            .iter()
121            .map(|record| record.meta().id())
122            .collect();
123        let table = crate::aggregate::table::<A>();
124        for (behavior, _) in &table.behaviors {
125            behavior.pruned(table.schema, table.name, &ids, plan);
126        }
127        #[cfg(feature = "search-engine")]
128        if let Some(index) = &self.index {
129            crate::search::queue_forget(index, &ids, plan)?;
130        }
131        for record in &self.records {
132            let meta = record.meta();
133            #[cfg(feature = "contracts")]
134            if let Some(name) = self.events.change {
135                let change = crate::change_event::pruned_change(meta);
136                plan.events.push(crate::RaisedEvent {
137                    name,
138                    payload: change,
139                });
140            }
141            if let Some(name) = self.events.pruned {
142                let pruned = crate::event::change_payload(
143                    serde_json::Map::new(),
144                    meta.id(),
145                    meta.version() + 1,
146                );
147                plan.events.push(crate::RaisedEvent {
148                    name,
149                    payload: serde_json::Value::Object(pruned),
150                });
151            }
152            #[cfg(feature = "cache")]
153            plan.announcements.changed(crate::ChangedRecord {
154                module: A::MODULE,
155                aggregate: A::NAME,
156                id: Some(record.meta().id()),
157            });
158            #[cfg(feature = "live")]
159            plan.announcements
160                .live_change(crate::TableChange::of::<A>(record.owner(), plan.reach()));
161        }
162        Ok(())
163    }
164}
165
166fn delete_loaded<A: Aggregate>(records: &[A], dialect: &dyn Dialect) -> Result<Step> {
167    let table = crate::aggregate::table::<A>().qualified(dialect);
168    let (id, version) = (dialect.quote("id"), dialect.quote("version"));
169    let expected = u64::try_from(records.len())
170        .map_err(|error| Error::internal("counting pruned records", error))?;
171    if dialect.binds_arrays() {
172        // One statement text whatever the batch size, so each connection prepares it once.
173        let sql = format!(
174            "DELETE FROM {table} WHERE ({id}, {version}) IN \
175             (SELECT * FROM unnest({}::uuid[], {}::bigint[]))",
176            dialect.placeholder(1),
177            dialect.placeholder(2)
178        );
179        let ids = records
180            .iter()
181            .map(|record| SqlValue::Uuid(record.meta().id()))
182            .collect();
183        let versions = records
184            .iter()
185            .map(|record| SqlValue::Int(record.meta().version()))
186            .collect();
187        let params = vec![SqlValue::Array(ids), SqlValue::Array(versions)];
188        return Ok(Step::statement(sql, params, Some(expected)));
189    }
190    let mut params = Vec::with_capacity(records.len() * 2);
191    let mut matches = Vec::with_capacity(records.len());
192    for record in records {
193        let meta = record.meta();
194        params.extend([SqlValue::Uuid(meta.id()), SqlValue::Int(meta.version())]);
195        matches.push(format!(
196            "({id} = {} AND {version} = {})",
197            dialect.placeholder(params.len() - 1),
198            dialect.placeholder(params.len())
199        ));
200    }
201    let sql = format!("DELETE FROM {table} WHERE {}", matches.join(" OR "));
202    Ok(Step::statement(sql, params, Some(expected)))
203}
sofie_db/prune.rs

sofie_db/
prune.rs
