eventuary_sqlite/
event_log.rs

1use rusqlite::Connection;
2
3use eventuary_core::Result;
4
5use crate::relation::SqliteRelationName;
6use crate::schema::{Migration, RelationReplacement};
7
8const EVENT_LOG_0001_INIT_SQL: &str = r#"
9CREATE TABLE IF NOT EXISTS {events} (
10    sequence INTEGER PRIMARY KEY AUTOINCREMENT,
11    id TEXT NOT NULL UNIQUE,
12    organization TEXT NOT NULL,
13    namespace TEXT NOT NULL,
14    topic TEXT NOT NULL,
15    event_key TEXT NOT NULL,
16    payload TEXT NOT NULL,
17    content_type TEXT NOT NULL,
18    metadata TEXT NOT NULL,
19    timestamp TEXT NOT NULL,
20    version INTEGER NOT NULL,
21    parent_id TEXT,
22    correlation_id TEXT,
23    causation_id TEXT,
24    partition_key TEXT,
25    partition_hash INTEGER,
26    partition_id INTEGER,
27    partition_count INTEGER,
28    partition_strategy TEXT
29);
30
31CREATE INDEX IF NOT EXISTS idx_events_org_seq ON {events} (organization, sequence);
32CREATE INDEX IF NOT EXISTS idx_events_org_ns_seq ON {events} (organization, namespace, sequence);
33CREATE INDEX IF NOT EXISTS idx_events_org_topic_seq ON {events} (organization, topic, sequence);
34CREATE INDEX IF NOT EXISTS idx_events_timestamp ON {events} (timestamp);
35
36CREATE INDEX IF NOT EXISTS idx_events_partition_count_id_sequence
37ON {events} (partition_count, partition_id, sequence)
38WHERE partition_id IS NOT NULL;
39
40CREATE INDEX IF NOT EXISTS idx_events_org_ns_partition_count_id_sequence
41ON {events} (organization, namespace, partition_count, partition_id, sequence)
42WHERE partition_id IS NOT NULL;
43"#;
44
45const EVENT_LOG_MIGRATIONS: &[Migration] = &[Migration {
46    name: "0001_init",
47    sql: EVENT_LOG_0001_INIT_SQL,
48}];
49
50#[derive(Debug, Clone)]
51pub struct SqliteEventLogSchemaConfig {
52    pub events_relation: SqliteRelationName,
53}
54
55impl Default for SqliteEventLogSchemaConfig {
56    fn default() -> Self {
57        Self {
58            events_relation: SqliteRelationName::new("events").expect("default events relation"),
59        }
60    }
61}
62
63pub struct SqliteEventLogSchema;
64
65impl SqliteEventLogSchema {
66    pub fn schema_sql(config: &SqliteEventLogSchemaConfig) -> String {
67        crate::schema::render_schema_sql(EVENT_LOG_MIGRATIONS, &replacements(config))
68    }
69
70    pub fn prepare(conn: &Connection, config: &SqliteEventLogSchemaConfig) -> Result<()> {
71        crate::schema::apply_schema(conn, EVENT_LOG_MIGRATIONS, &replacements(config))
72    }
73}
74
75fn replacements(config: &SqliteEventLogSchemaConfig) -> [RelationReplacement<'_>; 1] {
76    [RelationReplacement {
77        token: "{events}",
78        relation: &config.events_relation,
79    }]
80}
81
82#[cfg(test)]
83mod tests {
84    use super::*;
85
86    #[test]
87    fn schema_orders_required_key_after_topic() {
88        let sql = SqliteEventLogSchema::schema_sql(&SqliteEventLogSchemaConfig::default());
89        let normalized = sql
90            .replace(',', " ")
91            .split_whitespace()
92            .collect::<Vec<_>>()
93            .join(" ");
94        let id = normalized.find("id TEXT NOT NULL UNIQUE").unwrap();
95        let organization = normalized.find("organization TEXT NOT NULL").unwrap();
96        let namespace = normalized.find("namespace TEXT NOT NULL").unwrap();
97        let topic = normalized.find("topic TEXT NOT NULL").unwrap();
98        let key = normalized.find("event_key TEXT NOT NULL").unwrap();
99        let payload = normalized.find("payload TEXT NOT NULL").unwrap();
100        assert!(id < organization);
101        assert!(organization < namespace);
102        assert!(namespace < topic);
103        assert!(topic < key);
104        assert!(key < payload);
105    }
106
107    #[test]
108    fn schema_sql_contains_events_table() {
109        let sql = SqliteEventLogSchema::schema_sql(&SqliteEventLogSchemaConfig::default());
110        assert!(sql.contains("CREATE TABLE IF NOT EXISTS \"events\""));
111    }
112}
eventuary_sqlite/event_log.rs

eventuary_sqlite/
event_log.rs
