mithril_aggregator/event_store/database/
query.rs

1use crate::event_store::{Event, EventMessage};
2use chrono::Utc;
3use mithril_common::StdResult;
4use mithril_persistence::sqlite::{Query, SourceAlias, SqLiteEntity, WhereCondition};
5use serde_json::json;
6
7/// Queries to insert [EventMessage] to the sqlite database.
8pub struct InsertEventQuery {
9    condition: WhereCondition,
10}
11
12impl InsertEventQuery {
13    pub fn one(message: EventMessage) -> StdResult<Self> {
14        let condition = WhereCondition::new(
15            "(source, action, content, created_at) values (?*, ?*, ?*, ?*)",
16            vec![
17                sqlite::Value::String(message.source),
18                sqlite::Value::String(message.action),
19                sqlite::Value::String(serde_json::to_string(&json!({
20                    "content": message.content,
21                    "headers": message.headers,
22                }))?),
23                sqlite::Value::String(Utc::now().to_rfc3339()),
24            ],
25        );
26
27        Ok(Self { condition })
28    }
29}
30
31impl Query for InsertEventQuery {
32    type Entity = Event;
33
34    fn filters(&self) -> WhereCondition {
35        self.condition.clone()
36    }
37
38    fn get_definition(&self, data: &str) -> String {
39        let projection = Self::Entity::get_projection().expand(SourceAlias::default());
40
41        format!(r#"insert into event {data} returning {projection}"#)
42    }
43}
44
45#[cfg(test)]
46mod tests {
47    use super::*;
48
49    #[test]
50    fn provider_sql() {
51        let message = EventMessage::new("source", "action", &"content".to_string(), Vec::new());
52
53        let (final_expression, parameters) =
54            InsertEventQuery::one(message).unwrap().filters().expand();
55
56        assert_eq!(
57            "(source, action, content, created_at) values (?1, ?2, ?3, ?4)".to_string(),
58            final_expression
59        );
60        assert_eq!(4, parameters.len());
61    }
62
63    #[test]
64    fn build_a_json_for_content_field_with_content_and_headers() {
65        #[derive(serde::Serialize)]
66        struct Content {
67            attr1: String,
68            attr2: i32,
69        }
70        let content = Content {
71            attr1: "content".to_string(),
72            attr2: 123,
73        };
74        let message = EventMessage::new("source", "action", &content, [("key", "value")].to_vec());
75
76        let (_, parameters) = InsertEventQuery::one(message).unwrap().filters().expand();
77
78        assert_eq!(
79            sqlite::Value::String(
80                r#"{"content":{"attr1":"content","attr2":123},"headers":{"key":"value"}}"#
81                    .to_string()
82            ),
83            parameters[2]
84        );
85    }
86}
mithril_aggregator/event_store/database/query.rs

mithril_aggregator/event_store/database/
query.rs
