postgres-es 0.5.0

A Postgres implementation of an event repository for cqrs-es.
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
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121

pub(crate) struct SqlQueryFactory {
    event_table: String,
    select_events: String,
    insert_event: String,
    all_events: String,
    insert_snapshot: String,
    update_snapshot: String,
    select_snapshot: String,
}

impl SqlQueryFactory {
    pub fn new(event_table: &str, snapshot_table: &str) -> Self {
        Self {
            event_table: event_table.to_string(),
            select_events: format!("
SELECT aggregate_type, aggregate_id, sequence, event_type, event_version, payload, metadata
  FROM {event_table}
  WHERE aggregate_type = $1 AND aggregate_id = $2
  ORDER BY sequence"),
            insert_event: format!("
INSERT INTO {event_table} (aggregate_type, aggregate_id, sequence, event_type, event_version, payload, metadata)
VALUES ($1, $2, $3, $4, $5, $6, $7)"),
            all_events: format!("
SELECT aggregate_type, aggregate_id, sequence, event_type, event_version, payload, metadata
  FROM {event_table}
  WHERE aggregate_type = $1
  ORDER BY sequence"),
            insert_snapshot: format!("
INSERT INTO {snapshot_table} (aggregate_type, aggregate_id, last_sequence, current_snapshot, payload)
VALUES ($1, $2, $3, $4, $5)"),
            update_snapshot: format!("
UPDATE {snapshot_table}
  SET last_sequence= $3 , payload= $6, current_snapshot= $4
  WHERE aggregate_type= $1 AND aggregate_id= $2 AND current_snapshot= $5"),
            select_snapshot: format!("
SELECT aggregate_type, aggregate_id, last_sequence, current_snapshot, payload
  FROM {snapshot_table}
  WHERE aggregate_type = $1 AND aggregate_id = $2")
        }
    }
    pub fn select_events(&self) -> &str {
        &self.select_events
    }
    pub fn insert_event(&self) -> &str {
        &self.insert_event
    }
    pub fn insert_snapshot(&self) -> &str {
        &self.insert_snapshot
    }
    pub fn update_snapshot(&self) -> &str {
        &self.update_snapshot
    }
    pub fn select_snapshot(&self) -> &str {
        &self.select_snapshot
    }
    pub fn all_events(&self) -> &str {
        &self.all_events
    }
    pub fn get_last_events(&self, last_sequence: usize) -> String {
        format!(
            "
SELECT aggregate_type, aggregate_id, sequence, event_type, event_version, payload, metadata
  FROM {}
  WHERE aggregate_type = $1 AND aggregate_id = $2 AND sequence > {}
  ORDER BY sequence",
            &self.event_table, last_sequence
        )
    }
}

#[test]
fn test_queries() {
    let query_factory = SqlQueryFactory::new("my_events", "my_snapshots");
    assert_eq!(
        query_factory.select_events(),
        "
SELECT aggregate_type, aggregate_id, sequence, event_type, event_version, payload, metadata
  FROM my_events
  WHERE aggregate_type = $1 AND aggregate_id = $2
  ORDER BY sequence"
    );
    assert_eq!(query_factory.insert_event(), "
INSERT INTO my_events (aggregate_type, aggregate_id, sequence, event_type, event_version, payload, metadata)
VALUES ($1, $2, $3, $4, $5, $6, $7)");
    assert_eq!(
        query_factory.all_events(),
        "
SELECT aggregate_type, aggregate_id, sequence, event_type, event_version, payload, metadata
  FROM my_events
  WHERE aggregate_type = $1
  ORDER BY sequence"
    );
    assert_eq!(
        query_factory.insert_snapshot(),
        "
INSERT INTO my_snapshots (aggregate_type, aggregate_id, last_sequence, current_snapshot, payload)
VALUES ($1, $2, $3, $4, $5)"
    );
    assert_eq!(
        query_factory.update_snapshot(),
        "
UPDATE my_snapshots
  SET last_sequence= $3 , payload= $6, current_snapshot= $4
  WHERE aggregate_type= $1 AND aggregate_id= $2 AND current_snapshot= $5"
    );
    assert_eq!(
        query_factory.select_snapshot(),
        "
SELECT aggregate_type, aggregate_id, last_sequence, current_snapshot, payload
  FROM my_snapshots
  WHERE aggregate_type = $1 AND aggregate_id = $2"
    );
    assert_eq!(
        query_factory.get_last_events(20),
        "
SELECT aggregate_type, aggregate_id, sequence, event_type, event_version, payload, metadata
  FROM my_events
  WHERE aggregate_type = $1 AND aggregate_id = $2 AND sequence > 20
  ORDER BY sequence"
    );
}