eventuary_postgres/
buffer_store.rs

1//! PostgreSQL [`BufferStore`] implementation.
2//!
3//! Persists buffered events plus their cursor as JSONB. The generic
4//! cursor `C` must round-trip via `serde_json`. `pending` returns the
5//! current snapshot ordered by id; `nack` is a no-op (entries remain
6//! visible to the next `pending` call until acked).
7
8use std::marker::PhantomData;
9use std::sync::Arc;
10
11use serde::{Serialize, de::DeserializeOwned};
12use sqlx::{PgPool, Row};
13
14use eventuary_core::io::reader::{BufferEntry, BufferStore};
15use eventuary_core::{Error, Event, Result, SerializedEvent};
16
17use crate::relation::PgRelationName;
18use crate::schema::{Migration, RelationReplacement};
19
20const BUFFER_STORE_0001_INIT_SQL: &str = r#"
21CREATE TABLE IF NOT EXISTS {buffer_entries} (
22    id        BIGSERIAL    PRIMARY KEY,
23    event     JSONB        NOT NULL,
24    cursor    JSONB        NOT NULL,
25    pushed_at TIMESTAMPTZ NOT NULL DEFAULT NOW()
26);
27
28CREATE INDEX IF NOT EXISTS idx_buffer_entries_pushed_at ON {buffer_entries} (pushed_at);
29"#;
30
31const BUFFER_STORE_MIGRATIONS: &[Migration] = &[Migration {
32    name: "0001_init",
33    sql: BUFFER_STORE_0001_INIT_SQL,
34}];
35
36#[derive(Debug, Clone)]
37pub struct PgBufferStoreConfig {
38    pub relation: PgRelationName,
39}
40
41impl Default for PgBufferStoreConfig {
42    fn default() -> Self {
43        Self {
44            relation: PgRelationName::new("buffer_entries").expect("default buffer relation"),
45        }
46    }
47}
48
49#[derive(Debug, Clone, Copy, Eq, PartialEq, Hash)]
50pub struct PgBufferStoreId(i64);
51
52impl PgBufferStoreId {
53    pub fn as_i64(&self) -> i64 {
54        self.0
55    }
56}
57
58pub struct PgBufferStore<C> {
59    pool: PgPool,
60    relation: Arc<String>,
61    _cursor: PhantomData<C>,
62}
63
64impl<C> Clone for PgBufferStore<C> {
65    fn clone(&self) -> Self {
66        Self {
67            pool: self.pool.clone(),
68            relation: Arc::clone(&self.relation),
69            _cursor: PhantomData,
70        }
71    }
72}
73
74impl<C> PgBufferStore<C> {
75    pub fn new(pool: PgPool, config: PgBufferStoreConfig) -> Self {
76        Self {
77            pool,
78            relation: Arc::new(config.relation.render()),
79            _cursor: PhantomData,
80        }
81    }
82
83    pub async fn connect(pool: PgPool, config: PgBufferStoreConfig) -> Result<Self> {
84        Self::prepare_schema(&pool, &config).await?;
85        Ok(Self::new(pool, config))
86    }
87
88    pub async fn prepare_schema(pool: &PgPool, config: &PgBufferStoreConfig) -> Result<()> {
89        crate::schema::apply_schema(
90            pool,
91            BUFFER_STORE_MIGRATIONS,
92            &[RelationReplacement {
93                token: "{buffer_entries}",
94                relation: &config.relation,
95            }],
96        )
97        .await
98    }
99
100    pub fn schema_sql(config: &PgBufferStoreConfig) -> String {
101        crate::schema::render_schema_sql(
102            BUFFER_STORE_MIGRATIONS,
103            &[RelationReplacement {
104                token: "{buffer_entries}",
105                relation: &config.relation,
106            }],
107        )
108    }
109}
110
111fn encode_cursor<C: Serialize>(cursor: &C) -> Result<serde_json::Value> {
112    serde_json::to_value(cursor).map_err(|e| Error::Serialization(format!("buffer encode: {e}")))
113}
114
115fn decode_cursor<C: DeserializeOwned>(value: serde_json::Value) -> Result<C> {
116    serde_json::from_value(value)
117        .map_err(|e| Error::Serialization(format!("buffer decode cursor: {e}")))
118}
119
120fn encode_event(event: &Event) -> Result<serde_json::Value> {
121    let serialized = SerializedEvent::from_event(event)?;
122    serde_json::to_value(serialized)
123        .map_err(|e| Error::Serialization(format!("buffer encode event: {e}")))
124}
125
126fn decode_event(value: serde_json::Value) -> Result<Event> {
127    let serialized: SerializedEvent = serde_json::from_value(value)
128        .map_err(|e| Error::Serialization(format!("buffer decode event: {e}")))?;
129    serialized.to_event()
130}
131
132impl<C> BufferStore<C> for PgBufferStore<C>
133where
134    C: Serialize + DeserializeOwned + Clone + Send + Sync + 'static,
135{
136    type Id = PgBufferStoreId;
137
138    async fn push(&self, event: &Event, cursor: &C) -> Result<Self::Id> {
139        let sql = format!(
140            "INSERT INTO {relation} (event, cursor) VALUES ($1, $2) RETURNING id",
141            relation = self.relation
142        );
143        let row = sqlx::query(&sql)
144            .bind(encode_event(event)?)
145            .bind(encode_cursor(cursor)?)
146            .fetch_one(&self.pool)
147            .await
148            .map_err(|e| Error::Store(e.to_string()))?;
149        Ok(PgBufferStoreId(row.get::<i64, _>("id")))
150    }
151
152    async fn pending(&self) -> Result<Vec<BufferEntry<C, Self::Id>>> {
153        let sql = format!(
154            "SELECT id, event, cursor FROM {relation} ORDER BY id",
155            relation = self.relation
156        );
157        let rows = sqlx::query(&sql)
158            .fetch_all(&self.pool)
159            .await
160            .map_err(|e| Error::Store(e.to_string()))?;
161        let mut out = Vec::with_capacity(rows.len());
162        for row in rows {
163            let id: i64 = row.get("id");
164            let event_value: serde_json::Value = row.get("event");
165            let cursor_value: serde_json::Value = row.get("cursor");
166            out.push(BufferEntry {
167                id: PgBufferStoreId(id),
168                event: decode_event(event_value)?,
169                cursor: decode_cursor::<C>(cursor_value)?,
170            });
171        }
172        Ok(out)
173    }
174
175    async fn ack(&self, id: &Self::Id) -> Result<()> {
176        let sql = format!(
177            "DELETE FROM {relation} WHERE id = $1",
178            relation = self.relation
179        );
180        sqlx::query(&sql)
181            .bind(id.0)
182            .execute(&self.pool)
183            .await
184            .map_err(|e| Error::Store(e.to_string()))?;
185        Ok(())
186    }
187
188    async fn nack(&self, _id: &Self::Id) -> Result<()> {
189        Ok(())
190    }
191}
192
193#[cfg(test)]
194mod schema_tests {
195    use super::*;
196
197    #[test]
198    fn schema_sql_contains_expected_table() {
199        let sql =
200            PgBufferStore::<crate::reader::PgCursor>::schema_sql(&PgBufferStoreConfig::default());
201        assert!(sql.contains("CREATE TABLE IF NOT EXISTS \"buffer_entries\""));
202    }
203}
eventuary_postgres/buffer_store.rs

eventuary_postgres/
buffer_store.rs
