eventuary_postgres/
dedupe.rs

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

eventuary_postgres/
dedupe.rs
