eventuary_postgres/
multiplexer.rs

1//! PostgreSQL [`MultiplexerStore`] implementation.
2//!
3//! Records `(event_id, subscriber_id)` completion rows in the
4//! configured relation. `mark_completed` uses
5//! `INSERT ... ON CONFLICT DO NOTHING` so concurrent or redelivered
6//! calls converge without conflict.
7
8use std::sync::Arc;
9
10use sqlx::PgPool;
11
12use eventuary_core::io::handler::{MultiplexerKey, MultiplexerStore};
13use eventuary_core::{Error, Result};
14
15use crate::relation::PgRelationName;
16use crate::schema::{Migration, RelationReplacement};
17
18const MULTIPLEXER_STORE_0001_INIT_SQL: &str = r#"
19CREATE TABLE IF NOT EXISTS {multiplexer_completions} (
20    event_id      UUID        NOT NULL,
21    subscriber_id TEXT        NOT NULL,
22    completed_at  TIMESTAMPTZ NOT NULL DEFAULT NOW(),
23    PRIMARY KEY (event_id, subscriber_id)
24);
25"#;
26
27const MULTIPLEXER_STORE_MIGRATIONS: &[Migration] = &[Migration {
28    name: "0001_init",
29    sql: MULTIPLEXER_STORE_0001_INIT_SQL,
30}];
31
32#[derive(Debug, Clone)]
33pub struct PgMultiplexerStoreConfig {
34    pub relation: PgRelationName,
35}
36
37impl Default for PgMultiplexerStoreConfig {
38    fn default() -> Self {
39        Self {
40            relation: PgRelationName::new("multiplexer_completions")
41                .expect("default multiplexer relation"),
42        }
43    }
44}
45
46#[derive(Clone)]
47pub struct PgMultiplexerStore {
48    pool: PgPool,
49    relation: Arc<String>,
50}
51
52impl PgMultiplexerStore {
53    pub fn new(pool: PgPool, config: PgMultiplexerStoreConfig) -> Self {
54        Self {
55            pool,
56            relation: Arc::new(config.relation.render()),
57        }
58    }
59
60    pub async fn connect(pool: PgPool, config: PgMultiplexerStoreConfig) -> Result<Self> {
61        Self::prepare_schema(&pool, &config).await?;
62        Ok(Self::new(pool, config))
63    }
64
65    pub async fn prepare_schema(pool: &PgPool, config: &PgMultiplexerStoreConfig) -> Result<()> {
66        crate::schema::apply_schema(
67            pool,
68            MULTIPLEXER_STORE_MIGRATIONS,
69            &[RelationReplacement {
70                token: "{multiplexer_completions}",
71                relation: &config.relation,
72            }],
73        )
74        .await
75    }
76
77    pub fn schema_sql(config: &PgMultiplexerStoreConfig) -> String {
78        crate::schema::render_schema_sql(
79            MULTIPLEXER_STORE_MIGRATIONS,
80            &[RelationReplacement {
81                token: "{multiplexer_completions}",
82                relation: &config.relation,
83            }],
84        )
85    }
86}
87
88impl MultiplexerStore for PgMultiplexerStore {
89    async fn is_completed(&self, key: &MultiplexerKey) -> Result<bool> {
90        let event_id = key.event_id.to_string();
91        let sql = format!(
92            "SELECT 1 FROM {relation} \
93             WHERE event_id = $1::uuid AND subscriber_id = $2",
94            relation = self.relation
95        );
96        let row = sqlx::query(&sql)
97            .bind(event_id)
98            .bind(key.subscriber_id.as_str())
99            .fetch_optional(&self.pool)
100            .await
101            .map_err(|e| Error::Store(e.to_string()))?;
102        Ok(row.is_some())
103    }
104
105    async fn mark_completed(&self, key: &MultiplexerKey) -> Result<()> {
106        let event_id = key.event_id.to_string();
107        let sql = format!(
108            "INSERT INTO {relation} (event_id, subscriber_id) \
109             VALUES ($1::uuid, $2) \
110             ON CONFLICT (event_id, subscriber_id) DO NOTHING",
111            relation = self.relation
112        );
113        sqlx::query(&sql)
114            .bind(event_id)
115            .bind(key.subscriber_id.as_str())
116            .execute(&self.pool)
117            .await
118            .map_err(|e| Error::Store(e.to_string()))?;
119        Ok(())
120    }
121}
122
123#[cfg(test)]
124mod schema_tests {
125    use super::*;
126
127    #[test]
128    fn schema_sql_contains_expected_table() {
129        let sql = PgMultiplexerStore::schema_sql(&PgMultiplexerStoreConfig::default());
130        assert!(sql.contains("CREATE TABLE IF NOT EXISTS \"multiplexer_completions\""));
131    }
132}
eventuary_postgres/multiplexer.rs

eventuary_postgres/
multiplexer.rs
