sofie_db/
children.rs

1pub(crate) mod through;
2
3use std::ops::{Deref, DerefMut};
4
5use serde_json::Value;
6use sofie_core::encoding::hex;
7use sofie_core::{ColumnReader, ColumnSpec, Error, Result, RowReader, SqlType, SqlValue, Uuid};
8
9use crate::aggregate::{Table, field_columns};
10use crate::selection::Selection;
11use crate::sql::{equals, same_values};
12use crate::{Aggregate, Dialect, Field, Plan, Selected, Step};
13pub use through::{owned_through, refuse_parent_beyond_reach};
14
15#[derive(Debug)]
16pub struct ChildTable {
17    pub module: &'static str,
18    pub name: &'static str,
19    pub fields: &'static [Field],
20}
21
22pub trait Child: Sized + Send + Sync + 'static {
23    const TABLE: ChildTable;
24
25    fn id(&self) -> Uuid;
26
27    fn encode(&self, out: &mut Vec<SqlValue>);
28
29    fn decode(id: Uuid, row: &mut dyn ColumnReader) -> Result<Self>;
30
31    fn refuse_unlisted_moves(&self, loaded: &mut dyn ColumnReader) -> Result<()> {
32        let _ = loaded;
33        Ok(())
34    }
35}
36
37pub struct Children<T> {
38    rows: Vec<T>,
39    loaded: Vec<(Uuid, Vec<SqlValue>)>,
40}
41
42impl<T> Default for Children<T> {
43    fn default() -> Self {
44        Self {
45            rows: Vec::new(),
46            loaded: Vec::new(),
47        }
48    }
49}
50
51impl<T: Child> Children<T> {
52    pub fn push(&mut self, child: T) {
53        self.rows.push(child);
54    }
55
56    pub fn retain(&mut self, keep: impl FnMut(&T) -> bool) {
57        self.rows.retain(keep);
58    }
59
60    pub fn decode(row: &mut dyn ColumnReader) -> Result<Self> {
61        let Value::Array(rows) = row.json()? else {
62            return Err(Error::internal_message(format!(
63                "expected the rows of `{}` as a JSON array",
64                T::TABLE.name
65            )));
66        };
67        let loaded = rows
68            .into_iter()
69            .map(child_columns::<T>)
70            .collect::<Result<Vec<_>>>()?;
71        let rows = loaded
72            .iter()
73            .map(|(id, values)| T::decode(*id, &mut RowReader::new(values)))
74            .collect::<Result<_>>()?;
75        Ok(Self { rows, loaded })
76    }
77
78    fn current(&self) -> Vec<(Uuid, Vec<SqlValue>)> {
79        self.rows
80            .iter()
81            .map(|child| {
82                let mut values = Vec::new();
83                child.encode(&mut values);
84                (child.id(), values)
85            })
86            .collect()
87    }
88}
89
90impl<T> Deref for Children<T> {
91    type Target = [T];
92
93    fn deref(&self) -> &[T] {
94        &self.rows
95    }
96}
97
98impl<T> DerefMut for Children<T> {
99    fn deref_mut(&mut self) -> &mut [T] {
100        &mut self.rows
101    }
102}
103
104pub trait ChildRows: Send + Sync {
105    fn changed(&self) -> bool;
106
107    fn steps(&self, parent: &str, parent_id: Uuid, dialect: &dyn Dialect) -> Vec<Step>;
108
109    fn committed(&mut self);
110
111    fn refuse_unlisted_moves(&self) -> Result<()>;
112}
113
114impl<T: Child> ChildRows for Children<T> {
115    fn changed(&self) -> bool {
116        let current = self.current();
117        current.len() != self.loaded.len()
118            || current
119                .iter()
120                .zip(&self.loaded)
121                .any(|((id, values), (loaded, old))| id != loaded || !same_values(old, values))
122    }
123
124    fn steps(&self, parent: &str, parent_id: Uuid, dialect: &dyn Dialect) -> Vec<Step> {
125        let current = self.current();
126        let rows = ChildStatements {
127            table: &T::TABLE,
128            parent,
129            parent_id,
130            dialect,
131        };
132        let loaded = |id: &Uuid| self.loaded.iter().find(|(loaded, _)| loaded == id);
133        let deletes = self
134            .loaded
135            .iter()
136            .filter(|(id, _)| !current.iter().any(|(kept, _)| kept == id))
137            .map(|(id, _)| rows.delete(*id));
138        let updates = current.iter().filter_map(|(id, values)| {
139            let (_, old) = loaded(id)?;
140            (!same_values(old, values)).then(|| rows.update(*id, values))
141        });
142        let inserts = current
143            .iter()
144            .filter(|(id, _)| loaded(id).is_none())
145            .map(|(id, values)| rows.insert(*id, values));
146        deletes.chain(updates).chain(inserts).collect()
147    }
148
149    fn committed(&mut self) {
150        self.loaded = self.current();
151    }
152
153    fn refuse_unlisted_moves(&self) -> Result<()> {
154        for child in &self.rows {
155            let id = child.id();
156            if let Some((_, loaded)) = self.loaded.iter().find(|(loaded, _)| *loaded == id) {
157                child.refuse_unlisted_moves(&mut RowReader::new(loaded))?;
158            }
159        }
160        Ok(())
161    }
162}
163
164/// After their parent, whose owner it checks.
165pub(crate) fn plan_children<A: Aggregate>(
166    aggregate: &A,
167    plan: &mut Plan<'_>,
168    owned_through_parent: Option<Uuid>,
169) {
170    let id = aggregate.meta().id();
171    for children in aggregate.children() {
172        for step in children.steps(A::NAME, id, plan.dialect()) {
173            plan.push(step);
174        }
175    }
176    if let Some(owner) = owned_through_parent {
177        plan.push(through::ownership_check::<A>(id, owner, plan.dialect()));
178    }
179}
180
181/// Run before the write, so the row still holds the parent it was loaded with.
182pub(crate) fn check_loaded_parent<A: Aggregate>(
183    aggregate: &A,
184    plan: &mut Plan<'_>,
185    owned_through_parent: Option<Uuid>,
186) {
187    let meta = aggregate.meta();
188    if let (Some(owner), Some(_)) = (owned_through_parent, meta.snapshot.as_ref()) {
189        plan.push(through::ownership_check::<A>(
190            meta.id(),
191            owner,
192            plan.dialect(),
193        ));
194    }
195}
196
197pub(crate) fn committed<A: Aggregate>(aggregate: &mut A) {
198    for children in aggregate.children_mut() {
199        children.committed();
200    }
201}
202
203pub(crate) fn refuse_unlisted_moves<A: Aggregate>(aggregate: &A) -> Result<()> {
204    aggregate
205        .children()
206        .iter()
207        .try_for_each(|children| children.refuse_unlisted_moves())
208}
209
210pub(crate) fn children_changed<A: Aggregate>(aggregate: &A) -> bool {
211    aggregate
212        .children()
213        .iter()
214        .any(|children| children.changed())
215}
216
217struct ChildStatements<'a> {
218    table: &'a ChildTable,
219    parent: &'a str,
220    parent_id: Uuid,
221    dialect: &'a dyn Dialect,
222}
223
224impl ChildStatements<'_> {
225    fn qualified(&self) -> String {
226        self.dialect.qualified(self.table.module, self.table.name)
227    }
228
229    fn insert(&self, id: Uuid, values: &[SqlValue]) -> Step {
230        let mut columns = vec!["id".to_owned(), self.parent.to_owned()];
231        columns.extend(field_columns(self.table.fields));
232        let mut params = vec![SqlValue::Uuid(id), SqlValue::Uuid(self.parent_id)];
233        params.extend_from_slice(values);
234        let sql = crate::sql::insert(self.dialect, &self.qualified(), &columns);
235        Step::statement(sql, params, None)
236    }
237
238    /// Every column, so each table's update is one prepared statement.
239    fn update(&self, id: Uuid, values: &[SqlValue]) -> Step {
240        let mut params = Vec::new();
241        let columns = field_columns(self.table.fields);
242        let assignments: Vec<String> = columns
243            .iter()
244            .zip(values)
245            .map(|(column, value)| equals(self.dialect, column, value.clone(), &mut params))
246            .collect();
247        let condition = self.this_row(id, &mut params);
248        let sql = format!(
249            "UPDATE {} SET {} WHERE {condition}",
250            self.qualified(),
251            assignments.join(", ")
252        );
253        Step::statement(sql, params, None)
254    }
255
256    fn delete(&self, id: Uuid) -> Step {
257        let mut params = Vec::new();
258        let condition = self.this_row(id, &mut params);
259        let sql = format!("DELETE FROM {} WHERE {condition}", self.qualified());
260        Step::statement(sql, params, None)
261    }
262
263    /// Within its parent, so a forged id cannot reach another parent's child.
264    fn this_row(&self, id: Uuid, params: &mut Vec<SqlValue>) -> String {
265        let id = equals(self.dialect, "id", SqlValue::Uuid(id), params);
266        let parent = equals(
267            self.dialect,
268            self.parent,
269            SqlValue::Uuid(self.parent_id),
270            params,
271        );
272        format!("{id} AND {parent}")
273    }
274}
275
276impl ChildTable {
277    fn specs(&self) -> impl Iterator<Item = &'static ColumnSpec> {
278        self.fields.iter().flat_map(|field| field.columns)
279    }
280
281    pub(crate) fn selected(&'static self) -> Selected {
282        Selected(Selection::Children(self))
283    }
284
285    pub(crate) fn render(&self, base: &Table, dialect: &dyn Dialect) -> String {
286        let fields = field_columns(self.fields).into_iter().zip(self.specs());
287        let columns: Vec<String> = std::iter::once(dialect.quote("id"))
288            .chain(fields.map(|(column, spec)| {
289                let quoted = dialect.quote(&column);
290                match spec.ty {
291                    SqlType::Bytes => dialect.bytes_in_json(&quoted),
292                    _ => quoted,
293                }
294            }))
295            .collect();
296        let from = format!(
297            "{} WHERE {} = {}.{}",
298            dialect.qualified(self.module, self.name),
299            dialect.quote(base.name),
300            base.qualified(dialect),
301            dialect.quote("id")
302        );
303        dialect.nested_rows(&columns.join(", "), &from)
304    }
305}
306
307fn child_columns<T: Child>(row: Value) -> Result<(Uuid, Vec<SqlValue>)> {
308    let Value::Array(values) = row else {
309        return Err(Error::internal_message(format!(
310            "expected a row of `{}` as a JSON array",
311            T::TABLE.name
312        )));
313    };
314    let mut values = values.into_iter();
315    let id = match values.next() {
316        Some(Value::String(id)) => id
317            .parse()
318            .map_err(|error| Error::internal("reading a child row's id", error))?,
319        _ => return Err(Error::internal_message("a child row without its id")),
320    };
321    let columns = T::TABLE
322        .specs()
323        .zip(values)
324        .map(|(spec, value)| sql_value(spec.ty, value))
325        .collect::<Result<_>>()?;
326    Ok((id, columns))
327}
328
329fn sql_value(ty: SqlType, value: Value) -> Result<SqlValue> {
330    let unreadable = |error: &dyn std::fmt::Display| {
331        Error::internal_message(format!("reading a {ty:?} child column: {error}"))
332    };
333    Ok(match (ty, value) {
334        (_, Value::Null) => SqlValue::Null,
335        (SqlType::Bool, Value::Bool(value)) => SqlValue::Bool(value),
336        // A database without booleans or a timestamp type writes them as numbers.
337        (SqlType::Bool, Value::Number(number)) => SqlValue::Bool(number.as_i64() != Some(0)),
338        (SqlType::Timestamp, Value::Number(number)) => SqlValue::Timestamp(
339            number
340                .as_i64()
341                .and_then(|micros| sofie_core::Timestamp::from_microsecond(micros).ok())
342                .ok_or_else(|| unreadable(&number))?,
343        ),
344        (SqlType::Int, Value::Number(number)) => {
345            SqlValue::Int(number.as_i64().ok_or_else(|| unreadable(&number))?)
346        }
347        (SqlType::Float, Value::Number(number)) => {
348            SqlValue::Float(number.as_f64().ok_or_else(|| unreadable(&number))?)
349        }
350        // Postgres's JSON has no number for these, so it writes them as text.
351        (SqlType::Float, Value::String(text)) => SqlValue::Float(match text.as_str() {
352            "NaN" => f64::NAN,
353            "Infinity" => f64::INFINITY,
354            "-Infinity" => f64::NEG_INFINITY,
355            _ => return Err(unreadable(&text)),
356        }),
357        (SqlType::Text, Value::String(text)) => SqlValue::Text(text),
358        (SqlType::Uuid, Value::String(text)) => {
359            SqlValue::Uuid(text.parse().map_err(|error| unreadable(&error))?)
360        }
361        (SqlType::Timestamp, Value::String(text)) => {
362            SqlValue::Timestamp(text.parse().map_err(|error| unreadable(&error))?)
363        }
364        (SqlType::Date, Value::String(text)) => {
365            SqlValue::Date(text.parse().map_err(|error| unreadable(&error))?)
366        }
367        (SqlType::DateTime, Value::String(text)) => {
368            SqlValue::DateTime(text.parse().map_err(|error| unreadable(&error))?)
369        }
370        (SqlType::Bytes, Value::String(text)) => {
371            SqlValue::Bytes(hex_bytes(&text).ok_or_else(|| unreadable(&"bad hex"))?)
372        }
373        (_, other) => return Err(unreadable(&other)),
374    })
375}
376
377/// `\x0a1b`, the text form of a byte column.
378fn hex_bytes(text: &str) -> Option<Vec<u8>> {
379    hex::decode(text.strip_prefix("\\x")?)
380}
381
382#[cfg(test)]
383mod tests {
384    use super::*;
385
386    #[test]
387    fn json_columns_read_back_as_their_types() {
388        let read = |ty, value: Value| sql_value(ty, value).unwrap();
389
390        assert_eq!(read(SqlType::Int, 7.into()), SqlValue::Int(7));
391        assert_eq!(
392            read(SqlType::Text, "a".into()),
393            SqlValue::Text("a".to_owned())
394        );
395        assert_eq!(
396            read(SqlType::Bytes, "\\x0aff".into()),
397            SqlValue::Bytes(vec![10, 255])
398        );
399        assert_eq!(read(SqlType::Int, Value::Null), SqlValue::Null);
400        assert!(matches!(
401            read(SqlType::Timestamp, "2026-10-11T09:30:00.5+00:00".into()),
402            SqlValue::Timestamp(_)
403        ));
404        assert!(sql_value(SqlType::Int, "7".into()).is_err());
405    }
406
407    #[test]
408    fn a_float_postgres_writes_as_text_reads_back() {
409        let read = |text: &str| sql_value(SqlType::Float, text.into());
410
411        assert!(matches!(read("NaN"), Ok(SqlValue::Float(value)) if value.is_nan()));
412        assert_eq!(read("Infinity").unwrap(), SqlValue::Float(f64::INFINITY));
413        assert_eq!(
414            read("-Infinity").unwrap(),
415            SqlValue::Float(f64::NEG_INFINITY)
416        );
417        assert!(read("1.5").is_err());
418    }
419
420    struct Weight {
421        id: Uuid,
422        kilograms: f64,
423    }
424
425    impl Child for Weight {
426        const TABLE: ChildTable = ChildTable {
427            module: "stock",
428            name: "weight",
429            fields: &[Field::new(
430                "kilograms",
431                <f64 as sofie_core::Cast>::COLUMNS,
432                false,
433            )],
434        };
435
436        fn id(&self) -> Uuid {
437            self.id
438        }
439
440        fn encode(&self, out: &mut Vec<SqlValue>) {
441            out.push(SqlValue::Float(self.kilograms));
442        }
443
444        fn decode(id: Uuid, row: &mut dyn ColumnReader) -> Result<Self> {
445            Ok(Self {
446                id,
447                kilograms: row.float()?,
448            })
449        }
450    }
451
452    #[test]
453    fn a_child_float_changes_by_its_bits() {
454        let id = Uuid::now_v7().unwrap();
455        let dialect = crate::aggregate::tests::NUMBERED;
456        let mut children = Children::default();
457        children.push(Weight {
458            id,
459            kilograms: f64::NAN,
460        });
461        children.committed();
462
463        assert!(!children.changed());
464        assert!(children.steps("parcel", id, &dialect).is_empty());
465
466        children[0].kilograms = -0.0;
467        children.committed();
468        children[0].kilograms = 0.0;
469
470        assert!(children.changed());
471        assert_eq!(children.steps("parcel", id, &dialect).len(), 1);
472    }
473
474    struct Label {
475        id: Uuid,
476        text: String,
477        color: String,
478    }
479
480    impl Child for Label {
481        const TABLE: ChildTable = ChildTable {
482            module: "stock",
483            name: "label",
484            fields: &[
485                Field::new("text", <String as sofie_core::Cast>::COLUMNS, false),
486                Field::new("color", <String as sofie_core::Cast>::COLUMNS, false),
487            ],
488        };
489
490        fn id(&self) -> Uuid {
491            self.id
492        }
493
494        fn encode(&self, out: &mut Vec<SqlValue>) {
495            out.push(SqlValue::Text(self.text.clone()));
496            out.push(SqlValue::Text(self.color.clone()));
497        }
498
499        fn decode(id: Uuid, row: &mut dyn ColumnReader) -> Result<Self> {
500            Ok(Self {
501                id,
502                text: row.text()?.to_owned(),
503                color: row.text()?.to_owned(),
504            })
505        }
506    }
507
508    #[test]
509    fn a_child_row_is_updated_by_one_statement_whatever_changed() {
510        let dialect = crate::aggregate::tests::NUMBERED;
511        let mut labels = Children::default();
512        for _ in 0..2 {
513            labels.push(Label {
514                id: Uuid::now_v7().unwrap(),
515                text: "fragile".to_owned(),
516                color: "red".to_owned(),
517            });
518        }
519        labels.committed();
520
521        labels[0].text = "heavy".to_owned();
522        labels[1].color = "blue".to_owned();
523        let statements: Vec<String> = labels
524            .steps("parcel", Uuid::now_v7().unwrap(), &dialect)
525            .into_iter()
526            .map(|step| match step {
527                Step::Statement { sql, .. } | Step::Script(sql) => sql,
528            })
529            .collect();
530
531        assert_eq!(statements.len(), 2);
532        assert_eq!(statements[0], statements[1]);
533    }
534}
sofie_db/children.rs

sofie_db/
children.rs
