sofie_db/contracts/
mirror.rs

1use std::marker::PhantomData;
2use std::time::Duration;
3
4use serde::de::DeserializeOwned;
5use serde_json::{Map, Value, json};
6use sofie_core::{Error, Instant, Result, SqlValue, Uuid};
7
8use super::{Dep, Lookup};
9use crate::aggregate::table;
10use crate::change_event::DELETED_AT;
11use crate::event::PRUNED;
12use crate::worker::JobContext;
13use crate::{Aggregate, Context, Event, MessageKind, Outcome, Plan, Resolve};
14
15const PAGE: u64 = 500;
16
17pub trait Mirror: Aggregate {
18    /// The source's contract type, whose `Lookup` lists the records a new mirror starts from.
19    type Source: Send + Sync + 'static;
20    type Changed: Event;
21
22    fn columns(changed: &Map<String, Value>) -> Result<Vec<SqlValue>>;
23}
24
25pub fn changed_field<T: DeserializeOwned>(changed: &Map<String, Value>, name: &str) -> Result<T> {
26    let value = changed.get(name).cloned().unwrap_or(Value::Null);
27    serde_json::from_value(value).map_err(|error| {
28        Error::internal(format!("reading field `{name}` of a change event"), error)
29    })
30}
31
32pub struct MirrorRow<M> {
33    payload: Map<String, Value>,
34    mirror: PhantomData<fn() -> M>,
35}
36
37impl<M> MirrorRow<M> {
38    fn of(payload: Map<String, Value>) -> Self {
39        Self {
40            payload,
41            mirror: PhantomData,
42        }
43    }
44}
45
46/// Read untyped: a pruned record's event carries none of the fields `M::Changed` needs.
47impl<M: Mirror> Resolve<JobContext> for MirrorRow<M> {
48    fn resolve(context: &mut JobContext) -> impl Future<Output = Result<Self>> + Send {
49        let row = match context.event(<M::Changed as Event>::NAME) {
50            Ok(Value::Object(payload)) => Ok(Self::of(payload)),
51            Ok(_) => Err(Error::internal_message(
52                "a change event's payload is no object",
53            )),
54            Err(error) => Err(error),
55        };
56        std::future::ready(row)
57    }
58}
59
60/// A backfill locks the whole mirror instead, so the row lock is shared on the mirror.
61impl<M: Mirror> Outcome for MirrorRow<M> {
62    fn plan(&self, plan: &mut Plan<'_>) -> Result<()> {
63        let id: Uuid = changed_field(&self.payload, "id")?;
64        let dialect = plan.dialect();
65        if let (Some(shared), Some(exclusive)) = (dialect.shared_key_lock(), dialect.key_lock()) {
66            let key = format!("{}.{}:{id}", M::MODULE, M::NAME);
67            plan.push(crate::Step::statement(
68                shared,
69                vec![mirror_name::<M>()],
70                None,
71            ));
72            plan.push(crate::Step::statement(
73                exclusive,
74                vec![SqlValue::Text(key)],
75                None,
76            ));
77        }
78        self.write(id, plan)
79    }
80}
81
82/// Never overwrites a later version: change events may arrive out of order.
83impl<M: Mirror> MirrorRow<M> {
84    fn write(&self, id: Uuid, plan: &mut Plan<'_>) -> Result<()> {
85        let version: i64 = changed_field(&self.payload, "version")?;
86        let deleted_at: Option<Instant> = changed_field(&self.payload, DELETED_AT)?;
87        let table = table::<M>();
88        let hides = table.stored.iter().any(|column| column == DELETED_AT);
89        let pruned = changed_field::<Option<bool>>(&self.payload, PRUNED)? == Some(true);
90        if pruned || (deleted_at.is_some() && !hides) {
91            plan_forgetting::<M>(id, version, plan);
92            return Ok(());
93        }
94        let now = plan.stamp().now;
95        let mut params = vec![SqlValue::Uuid(id)];
96        params.extend(table.stored.iter().map(|column| match column.as_str() {
97            "version" => SqlValue::Int(version),
98            "created_at" | "updated_at" => SqlValue::Timestamp(now.timestamp()),
99            DELETED_AT => {
100                deleted_at.map_or(SqlValue::Null, |at| SqlValue::Timestamp(at.timestamp()))
101            }
102            _ => SqlValue::Null,
103        }));
104        params.extend(M::columns(&self.payload)?);
105        let dialect = plan.dialect();
106        let columns: Vec<&str> = table.row_columns().collect();
107        let updates: Vec<String> = columns[1..]
108            .iter()
109            .filter(|column| **column != "created_at")
110            .map(|column| format!("{0} = excluded.{0}", dialect.quote(column)))
111            .collect();
112        let qualified = table.qualified(dialect);
113        let sql = format!(
114            "{} ON CONFLICT (id) DO UPDATE SET {} WHERE {qualified}.{version} < excluded.{version}",
115            crate::sql::insert(dialect, &qualified, &columns),
116            updates.join(", "),
117            version = dialect.quote("version"),
118        );
119        plan.push(crate::Step::statement(sql, params, None));
120        plan.push(unforgotten_dropped::<M>(id, plan.dialect()));
121        announce::<M>(id, plan);
122        Ok(())
123    }
124}
125
126fn mirror_name<M: Mirror>() -> SqlValue {
127    SqlValue::Text(format!("{}.{}", M::MODULE, M::NAME))
128}
129
130/// An older change arriving after its record was dropped is taken back out at once.
131fn unforgotten_dropped<M: Mirror>(id: Uuid, dialect: &dyn crate::Dialect) -> crate::Step {
132    let sql = format!(
133        "DELETE FROM {table} WHERE {id} = {first} AND {version} <= \
134         (SELECT {version} FROM {forgotten} WHERE mirror = {second} AND {id} = {first})",
135        table = table::<M>().qualified(dialect),
136        forgotten = dialect.qualified("sofie", "mirror_forgotten"),
137        id = dialect.quote("id"),
138        version = dialect.quote("version"),
139        first = dialect.placeholder(1),
140        second = dialect.placeholder(2),
141    );
142    crate::Step::statement(sql, vec![SqlValue::Uuid(id), mirror_name::<M>()], None)
143}
144
145/// A pruned record, or a deleted one a mirror cannot hide, is dropped: no copy outlives its source.
146fn plan_forgetting<M: Mirror>(id: Uuid, version: i64, plan: &mut Plan<'_>) {
147    let dialect = plan.dialect();
148    let sql = format!(
149        "DELETE FROM {} WHERE {} = {} AND {} < {}",
150        table::<M>().qualified(dialect),
151        dialect.quote("id"),
152        dialect.placeholder(1),
153        dialect.quote("version"),
154        dialect.placeholder(2)
155    );
156    plan.push(crate::Step::statement(
157        sql,
158        vec![SqlValue::Uuid(id), SqlValue::Int(version)],
159        None,
160    ));
161    let forgotten = dialect.qualified("sofie", "mirror_forgotten");
162    let remembered = format!(
163        "INSERT INTO {forgotten} (mirror, {id}, {version}, forgotten_at) VALUES ({}, {}, {}, {}) \
164         ON CONFLICT (mirror, {id}) DO UPDATE SET {version} = excluded.{version}, \
165         forgotten_at = excluded.forgotten_at WHERE {forgotten}.{version} < excluded.{version}",
166        dialect.placeholder(1),
167        dialect.placeholder(2),
168        dialect.placeholder(3),
169        dialect.placeholder(4),
170        id = dialect.quote("id"),
171        version = dialect.quote("version"),
172    );
173    let params = vec![
174        mirror_name::<M>(),
175        SqlValue::Uuid(id),
176        SqlValue::Int(version),
177        SqlValue::Timestamp(plan.stamp().now.timestamp()),
178    ];
179    plan.push(crate::Step::statement(remembered, params, None));
180    announce::<M>(id, plan);
181}
182
183#[cfg_attr(not(feature = "cache"), expect(unused_variables))]
184fn announce<M: Mirror>(id: Uuid, plan: &mut Plan<'_>) {
185    #[cfg(feature = "cache")]
186    plan.changed(crate::ChangedRecord {
187        module: M::MODULE,
188        aggregate: M::NAME,
189        id: Some(id),
190    });
191    #[cfg(feature = "live")]
192    plan.live_change(crate::live::TableChange::of::<M>(None, plan.reach()));
193}
194
195/// Past the longest a late change event is retried for, a forgotten record cannot come back.
196const FORGOTTEN_FOR: Duration = Duration::from_hours(30 * 24);
197pub(crate) const SWEEP_FORGOTTEN: &str = "sofie.mirror.sweep_forgotten";
198
199pub(crate) async fn sweep_forgotten(swept: SweptForgotten) -> Result<SweptForgotten> {
200    Ok(swept)
201}
202
203pub(crate) struct SweptForgotten {
204    forgotten_for: Duration,
205}
206
207impl Resolve<JobContext> for SweptForgotten {
208    fn resolve(context: &mut JobContext) -> impl Future<Output = Result<Self>> + Send {
209        std::future::ready(Ok(Self {
210            forgotten_for: FORGOTTEN_FOR.max(context.retried_for),
211        }))
212    }
213}
214
215impl Outcome for SweptForgotten {
216    fn plan(&self, plan: &mut Plan<'_>) -> Result<()> {
217        let before = plan.stamp().now.checked_sub(self.forgotten_for)?;
218        let dialect = plan.dialect();
219        let sql = format!(
220            "DELETE FROM {} WHERE forgotten_at < {}",
221            dialect.qualified("sofie", "mirror_forgotten"),
222            dialect.placeholder(1)
223        );
224        plan.push(crate::Step::statement(
225            sql,
226            vec![SqlValue::Timestamp(before.timestamp())],
227            None,
228        ));
229        Ok(())
230    }
231}
232
233pub struct MirrorPage {
234    after: Option<Uuid>,
235}
236
237impl Resolve<JobContext> for MirrorPage {
238    fn resolve(context: &mut JobContext) -> impl Future<Output = Result<Self>> + Send {
239        let after = changed_field::<Option<Uuid>>(
240            context.payload.as_object().unwrap_or(&Map::new()),
241            "after",
242        );
243        std::future::ready(after.map(|after| Self { after }))
244    }
245}
246
247pub async fn backfill_mirror<M: Mirror>(
248    source: Dep<dyn Lookup<M::Source>>,
249    page: MirrorPage,
250    job: &'static str,
251) -> Result<Backfilled<M>> {
252    let changes = source.changes(page.after, PAGE).await?;
253    let full = u64::try_from(changes.len()).is_ok_and(|copied| copied == PAGE);
254    let last = changes
255        .last()
256        .map(|change| changed_field::<Uuid>(change, "id"));
257    let next = match last {
258        Some(last) if full => Some(last?),
259        _ => None,
260    };
261    Ok(Backfilled {
262        rows: changes.into_iter().map(MirrorRow::of).collect(),
263        next: next.map(|after| (job, after)),
264    })
265}
266
267pub struct Backfilled<M> {
268    rows: Vec<MirrorRow<M>>,
269    next: Option<(&'static str, Uuid)>,
270}
271
272impl<M: Mirror> Outcome for Backfilled<M> {
273    fn plan(&self, plan: &mut Plan<'_>) -> Result<()> {
274        if let Some(lock) = plan.dialect().key_lock() {
275            plan.push(crate::Step::statement(lock, vec![mirror_name::<M>()], None));
276        }
277        for row in &self.rows {
278            row.write(changed_field(&row.payload, "id")?, plan)?;
279        }
280        if let Some((job, after)) = self.next {
281            plan.queue(MessageKind::Job, [(job, json!({ "after": after }))])?;
282        }
283        Ok(())
284    }
285}
sofie_db/contracts/mirror.rs

sofie_db/contracts/
mirror.rs
