sofie-db 0.4.0

The handler model (inputs, events, replies), aggregates, read models, the query builder, the unit of work, migrations, the worker and the database seam.
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95

use sofie_core::{Error, Result, SqlValue, Timestamp};

use crate::{BoxFuture, Db, Step, Wakeups, query_values};

/// `notify` is sent on every change.
pub struct ShieldSql {
    pub upsert: &'static str,
    pub delete: &'static str,
    pub active: &'static str,
    pub notify: &'static str,
}

#[derive(Clone, Debug, PartialEq, Eq)]
pub struct SharedBlock {
    pub key: String,
    pub rule: String,
    pub grade: String,
    pub until: Timestamp,
}

pub struct SharedBlocks<'d> {
    db: &'d Db,
}

impl<'d> SharedBlocks<'d> {
    #[must_use]
    pub fn of(db: &'d Db) -> Self {
        Self { db }
    }

    fn sql(&self) -> Result<&'static ShieldSql> {
        self.db
            .driver()
            .dialect()
            .shield()
            .ok_or_else(|| Error::internal_message("this database keeps no shared shield blocks"))
    }

    pub async fn share(&self, block: &SharedBlock) -> Result<()> {
        let sql = self.sql()?;
        let params = vec![
            SqlValue::Text(block.key.clone()),
            SqlValue::Text(block.rule.clone()),
            SqlValue::Text(block.grade.clone()),
            SqlValue::Timestamp(block.until),
        ];
        self.db
            .transaction(vec![Step::statement(sql.upsert, params, None), notify(sql)])
            .await
    }

    pub async fn lift(&self, key: &str) -> Result<()> {
        let sql = self.sql()?;
        let params = vec![SqlValue::Text(key.to_owned())];
        self.db
            .transaction(vec![Step::statement(sql.delete, params, None), notify(sql)])
            .await
    }

    pub async fn active(&self, now: Timestamp) -> Result<Vec<SharedBlock>> {
        let params = [SqlValue::Timestamp(now)];
        let rows = query_values(self.db.driver(), self.sql()?.active, &params).await?;
        rows.into_iter().map(shared_block).collect()
    }

    #[must_use]
    pub fn wakeups(&self) -> Option<BoxFuture<'d, Result<Box<dyn Wakeups>>>> {
        self.db.driver().shield_wakeups()
    }
}

fn notify(sql: &ShieldSql) -> Step {
    Step::statement(sql.notify, Vec::new(), None)
}

fn shared_block(row: Vec<SqlValue>) -> Result<SharedBlock> {
    match <[SqlValue; 4]>::try_from(row) {
        Ok(
            [
                SqlValue::Text(key),
                SqlValue::Text(rule),
                SqlValue::Text(grade),
                SqlValue::Timestamp(until),
            ],
        ) => Ok(SharedBlock {
            key,
            rule,
            grade,
            until,
        }),
        _ => Err(Error::internal_message(
            "a shared shield block is not `key, rule, grade, until`",
        )),
    }
}