sofie_db/
across.rs

1use std::fmt;
2use std::marker::PhantomData;
3use std::sync::Arc;
4use std::sync::atomic::{AtomicU8, Ordering};
5
6use sofie_core::{Error, FieldError, Result, SqlValue, Uuid};
7
8use crate::statement::{Criteria, ONE};
9use crate::{Aggregate, Condition, ExpectedRows, Id, Outcome, Plan, Read, Step, Unmet};
10
11/// Whether records match across every owner, without any of them leaving: no rows, fields or ids.
12pub struct Across<A: Aggregate> {
13    read: Read,
14    criteria: Criteria,
15    marker: PhantomData<fn() -> A>,
16}
17
18impl<A: Aggregate> Across<A> {
19    /// Only for the derive's crate-private `across`: called elsewhere, it escapes that limit.
20    #[doc(hidden)]
21    #[must_use]
22    pub fn for_derived_across_only(read: &Read) -> Self {
23        Self {
24            read: read.across_owners(),
25            criteria: Criteria::default(),
26            marker: PhantomData,
27        }
28    }
29
30    #[must_use]
31    pub fn where_(mut self, condition: Condition<A>) -> Self {
32        self.criteria.conditions.push(condition.kind);
33        self
34    }
35
36    pub async fn exists(&self) -> Result<bool> {
37        let matched = self.criteria.clone().exists::<A>(&self.read).await?;
38        emit(
39            A::MODULE,
40            A::NAME,
41            self.read.actor(),
42            CheckKind::Exists,
43            matched,
44        );
45        Ok(matched)
46    }
47
48    pub async fn count(&self) -> Result<u64> {
49        let count = self.criteria.clone().count::<A>(&self.read).await?;
50        emit(
51            A::MODULE,
52            A::NAME,
53            self.read.actor(),
54            CheckKind::Count,
55            count > 0,
56        );
57        Ok(count)
58    }
59
60    /// The same check, run again inside the commit after serializing on the aggregate.
61    #[must_use]
62    pub fn guard(self, refusal: FieldError) -> Guard<A> {
63        Guard {
64            criteria: self.criteria,
65            refusal,
66            key: format!("{}.{}", A::MODULE, A::NAME),
67            marker: PhantomData,
68        }
69    }
70}
71
72/// Refuses its commit with its field error when a matching record exists by then; replies nothing.
73pub struct Guard<A: Aggregate> {
74    criteria: Criteria,
75    refusal: FieldError,
76    key: String,
77    marker: PhantomData<fn() -> A>,
78}
79
80impl<A: Aggregate> Guard<A> {
81    #[must_use]
82    pub fn except(mut self, id: Id<A>) -> Self {
83        self.criteria.conditions.push(A::ID.ne(id).kind);
84        self
85    }
86
87    /// Any two commits whose records could clash must share the key.
88    #[must_use]
89    pub fn keyed(mut self, key: impl fmt::Display) -> Self {
90        self.key = format!("{}.{}:{key}", A::MODULE, A::NAME);
91        self
92    }
93}
94
95impl<A: Aggregate> Outcome for Guard<A> {
96    fn plan(&self, plan: &mut Plan<'_>) -> Result<()> {
97        let dialect = plan.dialect();
98        let mut steps = Vec::with_capacity(2);
99        if let Some(lock) = dialect.key_lock() {
100            steps.push(Step::statement(
101                lock,
102                vec![SqlValue::Text(self.key.clone())],
103                None,
104            ));
105        }
106        let criteria = Criteria {
107            limit: Some(1),
108            ..self.criteria.clone()
109        };
110        let (sql, params) = criteria.select_sql::<A>(dialect, None, ONE);
111        let check = Arc::new(GuardCheck {
112            error: self.refusal.clone(),
113            module: A::MODULE,
114            aggregate: A::NAME,
115            actor: plan.reach().actor(),
116            state: AtomicU8::new(UNCHECKED),
117        });
118        steps.push(Step::Statement {
119            sql,
120            params,
121            expect_rows: Some(ExpectedRows {
122                rows: 0,
123                otherwise: Unmet::Refused(Arc::clone(&check)),
124            }),
125        });
126        plan.guard(steps, check);
127        Ok(())
128    }
129}
130
131const UNCHECKED: u8 = 0;
132const CLEAR: u8 = 1;
133const MATCHED: u8 = 2;
134
135/// Keeps its outcome for the committing task to audit: a driver may check on another thread.
136#[derive(Debug)]
137pub struct GuardCheck {
138    error: FieldError,
139    module: &'static str,
140    aggregate: &'static str,
141    actor: Option<Uuid>,
142    state: AtomicU8,
143}
144
145impl GuardCheck {
146    pub(crate) fn checked(&self, matched: bool) -> Result<()> {
147        self.state
148            .store(if matched { MATCHED } else { CLEAR }, Ordering::Relaxed);
149        if matched {
150            return Err(Error::Invalid(vec![self.error.clone()]));
151        }
152        Ok(())
153    }
154
155    pub(crate) fn audit(&self) {
156        let matched = match self.state.load(Ordering::Relaxed) {
157            CLEAR => false,
158            MATCHED => true,
159            _ => return,
160        };
161        emit(
162            self.module,
163            self.aggregate,
164            self.actor,
165            CheckKind::Guard,
166            matched,
167        );
168    }
169}
170
171impl PartialEq for GuardCheck {
172    fn eq(&self, other: &Self) -> bool {
173        (&self.error, self.module, self.aggregate, self.actor)
174            == (&other.error, other.module, other.aggregate, other.actor)
175    }
176}
177
178impl Eq for GuardCheck {}
179
180#[derive(Clone, Copy)]
181enum CheckKind {
182    Exists,
183    Count,
184    Guard,
185}
186
187impl CheckKind {
188    const fn name(self) -> &'static str {
189        match self {
190            Self::Exists => "exists",
191            Self::Count => "count",
192            Self::Guard => "guard",
193        }
194    }
195}
196
197/// Never the conditions' values: they may be another owner's data.
198fn emit(
199    module: &'static str,
200    aggregate: &'static str,
201    actor: Option<Uuid>,
202    kind: CheckKind,
203    matched: bool,
204) {
205    tracing::info!(
206        module,
207        aggregate,
208        actor = actor.map(tracing::field::display),
209        kind = kind.name(),
210        outcome = if matched { "matched" } else { "clear" },
211        "cross_owner_check"
212    );
213}
214
215#[cfg(test)]
216mod tests {
217    use sofie_core::Instant;
218
219    use super::*;
220    use crate::aggregate::tests::{NUMBERED, Note};
221    use crate::{Column, Meta, Reach};
222
223    const TITLE: Column<Note, String> = Column::new("title");
224
225    fn guard() -> Guard<Note> {
226        Across::<Note> {
227            read: never_read(),
228            criteria: Criteria::default(),
229            marker: PhantomData,
230        }
231        .where_(TITLE.eq("taken".to_owned()))
232        .guard(FieldError::invalid("title", "is taken"))
233    }
234
235    fn never_read() -> Read {
236        Read::within(&crate::Db::new(NoDriver), Reach::All)
237    }
238
239    struct NoDriver;
240
241    impl crate::Driver for NoDriver {
242        fn dialect(&self) -> &dyn crate::Dialect {
243            &NUMBERED
244        }
245
246        fn query<'a>(
247            &'a self,
248            _sql: &'a str,
249            _params: &'a [SqlValue],
250            _rows: &'a mut dyn crate::RowSink,
251        ) -> crate::BoxFuture<'a, Result<()>> {
252            unimplemented!("a planning test reads nothing")
253        }
254
255        fn transaction(&self, _steps: Vec<Step>) -> crate::BoxFuture<'_, Result<()>> {
256            unimplemented!("a planning test commits nothing")
257        }
258    }
259
260    fn planned(outcome: &impl Outcome, reach: Reach, leading: Vec<Step>) -> Vec<Step> {
261        let now = Instant::from(sofie_core::Timestamp::from_second(1_000).unwrap());
262        let mut plan = Plan::new(&NUMBERED, reach, now, leading, None);
263        outcome.plan(&mut plan).unwrap();
264        plan.into_parts().unwrap().steps
265    }
266
267    fn sql(step: &Step) -> &str {
268        match step {
269            Step::Statement { sql, .. } | Step::Script(sql) => sql,
270        }
271    }
272
273    #[test]
274    fn a_guard_listed_last_is_checked_after_the_leading_steps_and_before_any_write() {
275        let note = Note {
276            meta: Meta::new().unwrap(),
277            title: "taken".to_owned(),
278        };
279        let claim = Step::statement("INSERT INTO claims", Vec::new(), Some(1));
280
281        let steps = planned(&(note, guard()), Reach::All, vec![claim]);
282
283        let order: Vec<&str> = steps.iter().map(sql).collect();
284        assert_eq!(order[0], "INSERT INTO claims");
285        assert!(
286            order[1].starts_with("SELECT 1 FROM \"notes\".\"note\""),
287            "{order:?}"
288        );
289        assert!(
290            order[2].starts_with("INSERT INTO \"notes\".\"note\""),
291            "{order:?}"
292        );
293    }
294
295    #[test]
296    fn a_guard_checks_every_owner_s_records_except_those_it_names() {
297        let own = Id::<Note>::from_uuid(Uuid::from_u128(3));
298
299        let steps = planned(
300            &guard().except(own),
301            Reach::OwnedBy(Uuid::from_u128(7)),
302            Vec::new(),
303        );
304
305        let [
306            Step::Statement {
307                sql,
308                params,
309                expect_rows: Some(expected),
310            },
311        ] = &steps[..]
312        else {
313            panic!(
314                "expected the check alone, without a lock where commits run one at a time: {steps:?}"
315            );
316        };
317        assert_eq!(
318            sql,
319            "SELECT 1 FROM \"notes\".\"note\" WHERE (\"deleted_at\" IS NULL) AND (\"title\" = $1) \
320             AND NOT (\"id\" = $2) LIMIT $3"
321        );
322        assert_eq!(params[1], SqlValue::Uuid(Uuid::from_u128(3)));
323        assert_eq!(expected.rows, 0);
324    }
325
326    #[test]
327    fn a_matching_row_refuses_with_the_guard_s_field_error_and_none_lets_it_pass() {
328        let steps = planned(&guard(), Reach::All, Vec::new());
329        let Step::Statement {
330            expect_rows: Some(expected),
331            ..
332        } = &steps[0]
333        else {
334            panic!("expected the check: {steps:?}");
335        };
336
337        let refused = expected.check(1);
338
339        assert!(
340            matches!(&refused, Err(Error::Invalid(fields))
341                if fields[0].field == "title" && fields[0].message == "is taken"),
342            "{refused:?}"
343        );
344        assert!(expected.check(0).is_ok());
345    }
346}
sofie_db/across.rs

sofie_db/
across.rs
