runifold_store_postgres/
effect.rs

1//! `PostgreSQL` write-ahead external-effect persistence.
2
3use postgres::error::SqlState;
4use runifold_core::{CapabilityId, EffectId};
5use runifold_effect::{EffectExecutorError, EffectExecutorErrorKind, EffectRecord, EffectStore};
6use serde_json::Value;
7
8use crate::PostgresConversationStore;
9
10impl EffectStore for PostgresConversationStore {
11    fn load(&self, id: EffectId) -> Result<Option<EffectRecord>, EffectExecutorError> {
12        let sql = format!(
13            "SELECT record_json FROM {}_effects WHERE effect_id = $1",
14            self.table()
15        );
16        self.blocking()
17            .execute(move |client| client.query_opt(&sql, &[&id.as_uuid()]))
18            .map_err(|_| effect_worker())?
19            .map_err(effect_storage)?
20            .map(|row| decode_record(row.get("record_json")))
21            .transpose()
22    }
23
24    fn find_by_idempotency(
25        &self,
26        capability_id: CapabilityId,
27        key: &str,
28    ) -> Result<Option<EffectRecord>, EffectExecutorError> {
29        let sql = format!(
30            "SELECT record_json FROM {}_effects \
31             WHERE capability_id = $1 AND idempotency_key = $2",
32            self.table()
33        );
34        let key = key.to_owned();
35        self.blocking()
36            .execute(move |client| client.query_opt(&sql, &[&capability_id.as_uuid(), &key]))
37            .map_err(|_| effect_worker())?
38            .map_err(effect_storage)?
39            .map(|row| decode_record(row.get("record_json")))
40            .transpose()
41    }
42
43    fn compare_and_swap(
44        &self,
45        record: &EffectRecord,
46        expected_revision: Option<u64>,
47    ) -> Result<(), EffectExecutorError> {
48        let revision = effect_i64(record.revision)?;
49        let expected = expected_revision.map(effect_i64).transpose()?;
50        let encoded = serde_json::to_value(record).map_err(|_| effect_protocol())?;
51        let effect_id = record.request.effect_id.as_uuid();
52        let capability_id = record.request.capability_id.as_uuid();
53        let idempotency_key = record.request.idempotency_key.clone();
54        let table = format!("{}_effects", self.table());
55        self.blocking()
56            .execute(move |client| -> Result<(), EffectExecutorError> {
57                let mut transaction = client.transaction().map_err(effect_storage)?;
58                if let Some(key) = idempotency_key.as_deref() {
59                    let owner = transaction
60                        .query_opt(
61                            &format!(
62                                "SELECT effect_id FROM {table} \
63                                 WHERE capability_id = $1 AND idempotency_key = $2"
64                            ),
65                            &[&capability_id, &key],
66                        )
67                        .map_err(effect_storage)?
68                        .map(|row| row.get::<_, uuid::Uuid>("effect_id"));
69                    if owner.is_some_and(|owner| owner != effect_id) {
70                        return Err(effect_idempotency_conflict());
71                    }
72                }
73
74                match expected {
75                    None if revision == 0 => {
76                        let changed = transaction
77                            .execute(
78                                &format!(
79                                    "INSERT INTO {table} \
80                                     (effect_id, capability_id, idempotency_key, revision, record_json) \
81                                     VALUES ($1, $2, $3, $4, $5) \
82                                     ON CONFLICT (effect_id) DO NOTHING"
83                                ),
84                                &[
85                                    &effect_id,
86                                    &capability_id,
87                                    &idempotency_key,
88                                    &revision,
89                                    &encoded,
90                                ],
91                            )
92                            .map_err(effect_write_error)?;
93                        if changed != 1 {
94                            return Err(effect_conflict());
95                        }
96                    }
97                    Some(expected)
98                        if expected
99                            .checked_add(1)
100                            .is_some_and(|next| revision == next) =>
101                    {
102                        let changed = transaction
103                            .execute(
104                                &format!(
105                                    "UPDATE {table} SET capability_id = $1, idempotency_key = $2, \
106                                        revision = $3, record_json = $4 \
107                                     WHERE effect_id = $5 AND revision = $6"
108                                ),
109                                &[
110                                    &capability_id,
111                                    &idempotency_key,
112                                    &revision,
113                                    &encoded,
114                                    &effect_id,
115                                    &expected,
116                                ],
117                            )
118                            .map_err(effect_write_error)?;
119                        if changed != 1 {
120                            return Err(effect_conflict());
121                        }
122                    }
123                    _ => return Err(effect_conflict()),
124                }
125                transaction.commit().map_err(effect_storage)
126            })
127            .map_err(|_| effect_worker())?
128    }
129}
130
131fn decode_record(value: Value) -> Result<EffectRecord, EffectExecutorError> {
132    serde_json::from_value(value).map_err(|_| effect_protocol())
133}
134
135fn effect_i64(value: u64) -> Result<i64, EffectExecutorError> {
136    i64::try_from(value).map_err(|_| {
137        EffectExecutorError::new(
138            EffectExecutorErrorKind::Protocol,
139            "effect revision exceeds PostgreSQL BIGINT",
140        )
141    })
142}
143
144fn effect_storage(_error: postgres::Error) -> EffectExecutorError {
145    EffectExecutorError::new(
146        EffectExecutorErrorKind::Store,
147        "PostgreSQL effect store operation failed",
148    )
149}
150
151fn effect_worker() -> EffectExecutorError {
152    EffectExecutorError::new(
153        EffectExecutorErrorKind::Store,
154        "PostgreSQL effect store worker is unavailable",
155    )
156}
157
158fn effect_write_error(error: postgres::Error) -> EffectExecutorError {
159    if error.code() == Some(&SqlState::UNIQUE_VIOLATION) {
160        effect_idempotency_conflict()
161    } else {
162        effect_storage(error)
163    }
164}
165
166fn effect_protocol() -> EffectExecutorError {
167    EffectExecutorError::new(
168        EffectExecutorErrorKind::Protocol,
169        "PostgreSQL effect record is invalid",
170    )
171}
172
173fn effect_conflict() -> EffectExecutorError {
174    EffectExecutorError::new(
175        EffectExecutorErrorKind::Store,
176        "effect record revision precondition failed",
177    )
178}
179
180fn effect_idempotency_conflict() -> EffectExecutorError {
181    EffectExecutorError::new(
182        EffectExecutorErrorKind::IdempotencyConflict,
183        "idempotency key already belongs to another effect",
184    )
185}
runifold_store_postgres/effect.rs

runifold_store_postgres/
effect.rs
