sofie_db/
history.rs

1use sofie_core::variants::{self, Variants};
2use sofie_core::{Cast, ColumnReader, Error, Instant, Result, SqlType, SqlValue, Uuid};
3
4use crate::behavior::stored;
5use crate::{
6    Aggregate, Behavior, Column, Dialect, Field, Id, Plan, Read, Select, Step, Uses, Write,
7};
8
9/// Null for a record written for no principal, such as by a job.
10pub struct Audited;
11
12impl Behavior for Audited {
13    fn columns(&self) -> &'static [Field] {
14        const {
15            &[
16                Field::new("created_by", Uuid::COLUMNS, true),
17                Field::new("updated_by", Uuid::COLUMNS, true),
18            ]
19        }
20    }
21
22    fn written(&self, write: &Write<'_>, values: &mut [SqlValue]) {
23        let actor = write.stamp.actor.map_or(SqlValue::Null, SqlValue::Uuid);
24        values[0] = write.loaded("created_by").unwrap_or(&actor).clone();
25        values[1] = actor;
26    }
27}
28
29pub trait Authored: Aggregate {
30    const CREATED_BY: Column<Self, Option<Uuid>> = Column::new("created_by");
31    const UPDATED_BY: Column<Self, Option<Uuid>> = Column::new("updated_by");
32
33    fn created_by(&self) -> Option<Uuid> {
34        principal::<Self>(self, "created_by")
35    }
36
37    fn updated_by(&self) -> Option<Uuid> {
38        principal::<Self>(self, "updated_by")
39    }
40}
41
42impl<A: Uses<Audited>> Authored for A {}
43
44fn principal<A: Aggregate>(aggregate: &A, column: &str) -> Option<Uuid> {
45    match stored(aggregate, column) {
46        Some(SqlValue::Uuid(principal)) => Some(*principal),
47        _ => None,
48    }
49}
50
51pub struct RecordsChanges;
52
53impl Behavior for RecordsChanges {
54    fn columns(&self) -> &'static [Field] {
55        &[]
56    }
57
58    fn written(&self, _write: &Write<'_>, _values: &mut [SqlValue]) {}
59
60    fn planned(&self, write: &Write<'_>, plan: &mut Plan<'_>) -> Result<()> {
61        let dialect = plan.dialect();
62        let fields: Vec<serde_json::Value> = write
63            .changed_fields()
64            .into_iter()
65            .map(serde_json::Value::from)
66            .collect();
67        let columns = [
68            "id",
69            "record_table",
70            "record",
71            "owner",
72            "change",
73            "fields",
74            "actor",
75            "action",
76            "changed_at",
77        ];
78        let params = vec![
79            SqlValue::Uuid(Uuid::now_v7()?),
80            SqlValue::Text(record_table(write.table.schema, write.table.name)),
81            SqlValue::Uuid(write.id),
82            write.owner.map_or(SqlValue::Null, SqlValue::Uuid),
83            SqlValue::Text(ChangeKind::of(write).name().to_owned()),
84            SqlValue::Json(fields.into()),
85            write.stamp.actor.map_or(SqlValue::Null, SqlValue::Uuid),
86            write
87                .stamp
88                .action
89                .map_or(SqlValue::Null, |action| SqlValue::Text(action.to_owned())),
90            SqlValue::Timestamp(write.stamp.now.timestamp()),
91        ];
92        let table = history_table(dialect);
93        let sql = crate::sql::insert(dialect, &table, &columns);
94        plan.push(Step::statement(sql, params, None));
95        Ok(())
96    }
97}
98
99fn history_table(dialect: &dyn Dialect) -> String {
100    dialect.qualified("sofie", "history")
101}
102
103fn record_table(schema: &str, name: &str) -> String {
104    format!("{schema}.{name}")
105}
106
107#[derive(Clone, Copy, Debug, PartialEq, Eq)]
108pub enum ChangeKind {
109    Created,
110    Updated,
111    Deleted,
112    Restored,
113}
114
115impl ChangeKind {
116    fn of(write: &Write<'_>) -> Self {
117        if write.loaded.is_none() {
118            return Self::Created;
119        }
120        match write.deleted() {
121            (false, true) => Self::Deleted,
122            (true, false) => Self::Restored,
123            _ => Self::Updated,
124        }
125    }
126}
127
128impl Variants for ChangeKind {
129    const TYPE: &'static str = "history change";
130    const VARIANTS: &'static [(&'static str, Self)] = &[
131        ("CREATED", Self::Created),
132        ("UPDATED", Self::Updated),
133        ("DELETED", Self::Deleted),
134        ("RESTORED", Self::Restored),
135    ];
136}
137
138#[derive(Clone, Debug, PartialEq)]
139pub struct Change {
140    pub kind: ChangeKind,
141    pub fields: Vec<String>,
142    pub actor: Option<Uuid>,
143    pub action: Option<String>,
144    pub at: Instant,
145}
146
147/// An owned area reads the history of a record its principal owns now, through a parent too.
148pub trait History: Uses<RecordsChanges> {
149    fn history(read: &Read, id: Id<Self>) -> impl Future<Output = Result<Vec<Change>>> + Send {
150        history_of::<Self>(read, id)
151    }
152}
153
154pub async fn history_of<A: Aggregate>(read: &Read, id: Id<A>) -> Result<Vec<Change>> {
155    if read.owner().is_some() && !reachable(read, id).await? {
156        return Ok(Vec::new());
157    }
158    let dialect = read.dialect();
159    let params = vec![
160        SqlValue::Text(record_table(A::MODULE, A::NAME)),
161        SqlValue::Uuid(id.uuid()),
162    ];
163    let sql = format!(
164        "SELECT change, fields, actor, action, changed_at FROM {} \
165         WHERE record_table = {} AND record = {} ORDER BY changed_at, id",
166        history_table(dialect),
167        dialect.placeholder(1),
168        dialect.placeholder(2)
169    );
170    #[cfg(feature = "live")]
171    crate::live::read_tables([(A::MODULE, A::NAME)]);
172    let types = vec![
173        Some(SqlType::Text),
174        None,
175        Some(SqlType::Uuid),
176        Some(SqlType::Text),
177        Some(SqlType::Timestamp),
178    ];
179    read.decoded(&sql, &params, 0, types, change).await
180}
181
182async fn reachable<A: Aggregate>(read: &Read, id: Id<A>) -> Result<bool> {
183    let mut query = A::query().where_(A::ID.eq(id));
184    query.criteria.with_hidden = true;
185    query.exists(read).await
186}
187
188impl<A: Uses<RecordsChanges>> History for A {}
189
190fn change(row: &mut dyn ColumnReader) -> Result<Change> {
191    let kind = variants::decode(row)?;
192    let serde_json::Value::Array(fields) = row.json()? else {
193        return Err(Error::internal_message(
194            "expected the changed fields as a JSON array",
195        ));
196    };
197    Ok(Change {
198        kind,
199        fields: fields
200            .into_iter()
201            .filter_map(|field| field.as_str().map(str::to_owned))
202            .collect(),
203        actor: Option::<Uuid>::decode(row)?,
204        action: Option::<String>::decode(row)?,
205        at: Instant::decode(row)?,
206    })
207}
sofie_db/history.rs

sofie_db/
history.rs
