turnframe_store_postgres/
events.rs

1//! The claim ledger over `tf_domain_event` (spec §17.1, ADR-012).
2//!
3//! Nothing here deletes, and nothing inserts except the batch insert below.
4//! `sequence` is an identity column, so the store assigns the position rather
5//! than the caller, and a read ordered by it returns events in the order they
6//! were committed.
7//!
8//! There is exactly one `UPDATE` in this module, [`redact_payload`], and it is
9//! how personal data leaves a ledger that may not lose an event. It writes
10//! `payload`, `redacted_at` and `redaction_authority` and nothing else, so the
11//! primary key, the identity column, the case columns and `occurred_at` are
12//! untouched by the statement itself rather than by a promise: an event cannot
13//! change position because no statement here can write its position.
14//!
15//! A batch is one statement. A duplicate identifier — whether it collides with
16//! a stored event or with another event of the same batch — violates the
17//! primary key, and the statement fails whole: a receipt can never be backed by
18//! half of a commit.
19
20use async_trait::async_trait;
21use chrono::{DateTime, Utc};
22use sqlx::PgConnection;
23use sqlx::postgres::PgRow;
24use turnframe_core::case::CaseKey;
25use turnframe_core::event::EventRedaction;
26use turnframe_core::ids::{AccountId, CaseRevision, CommandId, EventId, RedactionAuthority};
27use turnframe_store::error::{StoreError, invalid_record};
28use turnframe_store::events::{
29    EventBatch, EventCursor, EventJournalReader, EventJournalWriter, EventPage, StoredEvent,
30};
31use uuid::Uuid;
32
33use crate::codec::{
34    column, count_from_sql, cursor_to_sql, limit_to_sql, revision_from_sql, revision_to_sql,
35    sequence_from_sql,
36};
37use crate::error::store_error;
38use crate::store::{PgStores, commit};
39
40/// Everything a read needs to rebuild a [`StoredEvent`].
41const EVENT_COLUMNS: &str = "sequence, event_id, workflow_key, case_id, case_revision, \
42     command_id, event_type, payload, occurred_at, redacted_at, redaction_authority";
43
44/// Appends a batch and returns its identifiers in order.
45pub(crate) async fn append(
46    conn: &mut PgConnection,
47    batch: EventBatch,
48) -> Result<Vec<EventId>, StoreError> {
49    if batch.is_empty() {
50        return Err(invalid_record());
51    }
52    let ids: Vec<Uuid> = batch.events.iter().map(|e| *e.event_id.as_uuid()).collect();
53    let types: Vec<String> = batch.events.iter().map(|e| e.event_type.clone()).collect();
54    let payloads: Vec<serde_json::Value> = batch.events.iter().map(|e| e.payload.clone()).collect();
55    let occurred: Vec<DateTime<Utc>> = batch.events.iter().map(|e| e.occurred_at).collect();
56    // `WITH ORDINALITY` plus `ORDER BY` fixes the order rows are inserted in, so
57    // the identity column hands out positions in the batch's own order.
58    sqlx::query(
59        "INSERT INTO tf_domain_event (
60             account_id, event_id, workflow_key, case_id, case_revision, command_id,
61             event_type, payload, occurred_at
62         )
63         SELECT $1, e.event_id, $2, $3, $4, $5, e.event_type, e.payload, e.occurred_at
64         FROM unnest($6::uuid[], $7::text[], $8::jsonb[], $9::timestamptz[])
65              WITH ORDINALITY AS e(event_id, event_type, payload, occurred_at, position)
66         ORDER BY e.position",
67    )
68    .bind(batch.account_id.as_str())
69    .bind(batch.case_key.workflow.as_str())
70    .bind(batch.case_key.case_id.as_str())
71    .bind(revision_to_sql(batch.revision)?)
72    .bind(batch.command_id.as_uuid())
73    .bind(ids)
74    .bind(types)
75    .bind(payloads)
76    .bind(occurred)
77    .execute(conn)
78    .await
79    .map_err(|error| store_error(&error))?;
80    Ok(batch.event_ids())
81}
82
83/// The events of a case past `since`, in append order.
84pub(crate) async fn list_since(
85    conn: &mut PgConnection,
86    account: &AccountId,
87    case_key: &CaseKey,
88    since: CaseRevision,
89    limit: usize,
90) -> Result<Vec<StoredEvent>, StoreError> {
91    let statement = format!(
92        "SELECT {EVENT_COLUMNS} FROM tf_domain_event
93         WHERE account_id = $1 AND workflow_key = $2 AND case_id = $3 AND case_revision > $4
94         ORDER BY sequence
95         LIMIT $5"
96    );
97    let rows = sqlx::query(&statement)
98        .bind(account.as_str())
99        .bind(case_key.workflow.as_str())
100        .bind(case_key.case_id.as_str())
101        .bind(revision_to_sql(since)?)
102        .bind(limit_to_sql(limit))
103        .fetch_all(conn)
104        .await
105        .map_err(|error| store_error(&error))?;
106    rows.iter().map(|row| decode_event(row, account)).collect()
107}
108
109/// The account's next events after `after`, across every case.
110///
111/// The cursor is the store-assigned sequence, and the sequence is the journal's
112/// total order, so `sequence > cursor` is both the paging condition and the
113/// ordering: an append always lands beyond every cursor already issued, and a
114/// page can therefore never repeat or reorder what an earlier one delivered.
115/// Another tenant's events are excluded by the `WHERE` clause rather than
116/// filtered afterwards, so they never consume a page's limit.
117pub(crate) async fn read_from(
118    conn: &mut PgConnection,
119    account: &AccountId,
120    after: EventCursor,
121    limit: usize,
122) -> Result<EventPage, StoreError> {
123    let statement = format!(
124        "SELECT {EVENT_COLUMNS} FROM tf_domain_event
125         WHERE account_id = $1 AND sequence > $2
126         ORDER BY sequence
127         LIMIT $3"
128    );
129    let rows = sqlx::query(&statement)
130        .bind(account.as_str())
131        .bind(cursor_to_sql(after))
132        .bind(limit_to_sql(limit))
133        .fetch_all(conn)
134        .await
135        .map_err(|error| store_error(&error))?;
136    let events = rows
137        .iter()
138        .map(|row| decode_event(row, account))
139        .collect::<Result<Vec<StoredEvent>, StoreError>>()?;
140    Ok(EventPage::new(events, after))
141}
142
143/// The events with the given identifiers, in the order they were asked for.
144///
145/// Receipt verification reads the ledger back this way, so an identifier that
146/// belongs to another tenant is simply absent from the answer rather than an
147/// error: a receipt citing it fails to verify, which is the point.
148pub(crate) async fn get_by_ids(
149    conn: &mut PgConnection,
150    account: &AccountId,
151    ids: &[EventId],
152) -> Result<Vec<StoredEvent>, StoreError> {
153    if ids.is_empty() {
154        return Ok(Vec::new());
155    }
156    let wanted: Vec<Uuid> = ids.iter().map(|id| *id.as_uuid()).collect();
157    let statement = format!(
158        "SELECT {EVENT_COLUMNS} FROM tf_domain_event
159         WHERE account_id = $1 AND event_id = ANY($2)"
160    );
161    let rows = sqlx::query(&statement)
162        .bind(account.as_str())
163        .bind(wanted)
164        .fetch_all(conn)
165        .await
166        .map_err(|error| store_error(&error))?;
167    let found = rows
168        .iter()
169        .map(|row| decode_event(row, account))
170        .collect::<Result<Vec<StoredEvent>, StoreError>>()?;
171    Ok(ids
172        .iter()
173        .filter_map(|id| found.iter().find(|event| event.event_id == *id).cloned())
174        .collect())
175}
176
177/// Empties one event's payload and records the erasure on the same row.
178///
179/// One statement does both, so there is no window in which a payload is gone
180/// and nobody knows who removed it. `COALESCE` is what makes a repeated request
181/// a no-op that keeps the **first** record: an erasure retried by an operator
182/// or a queue must not rewrite the history of who erased what, and the second
183/// call has nothing left to erase. `RETURNING` then hands back the record the
184/// row now carries, whichever call wrote it.
185///
186/// The `WHERE` clause is account-scoped, so an identifier belonging to another
187/// tenant updates nothing and is reported as `NotFound`, exactly like one that
188/// never existed: a write is not a way to probe for a neighbour's data.
189pub(crate) async fn redact_payload(
190    conn: &mut PgConnection,
191    account: &AccountId,
192    event_id: &EventId,
193    authority: &RedactionAuthority,
194) -> Result<EventRedaction, StoreError> {
195    let row = sqlx::query(
196        "UPDATE tf_domain_event
197            SET payload = 'null'::jsonb,
198                redacted_at = COALESCE(redacted_at, now()),
199                redaction_authority = COALESCE(redaction_authority, $3)
200          WHERE account_id = $1 AND event_id = $2
201      RETURNING redacted_at, redaction_authority",
202    )
203    .bind(account.as_str())
204    .bind(event_id.as_uuid())
205    .bind(authority.as_str())
206    .fetch_optional(conn)
207    .await
208    .map_err(|error| store_error(&error))?
209    .ok_or(StoreError::NotFound)?;
210    decode_redaction(&row)?.ok_or(StoreError::Corrupt)
211}
212
213/// How many events a case has.
214pub(crate) async fn count(
215    conn: &mut PgConnection,
216    account: &AccountId,
217    case_key: &CaseKey,
218) -> Result<u64, StoreError> {
219    let row = sqlx::query(
220        "SELECT count(*) AS total FROM tf_domain_event
221         WHERE account_id = $1 AND workflow_key = $2 AND case_id = $3",
222    )
223    .bind(account.as_str())
224    .bind(case_key.workflow.as_str())
225    .bind(case_key.case_id.as_str())
226    .fetch_one(conn)
227    .await
228    .map_err(|error| store_error(&error))?;
229    count_from_sql(column(&row, "total")?)
230}
231
232/// Rebuilds the erasure record of a row, when it carries one.
233///
234/// The two columns are constrained to agree, so half a record is stored data
235/// violating an invariant the adapter relies on: `Corrupt`, not a guess.
236fn decode_redaction(row: &PgRow) -> Result<Option<EventRedaction>, StoreError> {
237    let redacted_at: Option<DateTime<Utc>> = column(row, "redacted_at")?;
238    let authority: Option<String> = column(row, "redaction_authority")?;
239    match (redacted_at, authority) {
240        (None, None) => Ok(None),
241        (Some(redacted_at), Some(authority)) => Ok(Some(EventRedaction {
242            redacted_at,
243            authority: RedactionAuthority::from(authority),
244        })),
245        _ => Err(StoreError::Corrupt),
246    }
247}
248
249/// Rebuilds a stored event from its row.
250fn decode_event(row: &PgRow, account: &AccountId) -> Result<StoredEvent, StoreError> {
251    let sequence: i64 = column(row, "sequence")?;
252    let event_id: Uuid = column(row, "event_id")?;
253    let workflow_key: String = column(row, "workflow_key")?;
254    let case_id: String = column(row, "case_id")?;
255    let revision: i64 = column(row, "case_revision")?;
256    let command_id: Uuid = column(row, "command_id")?;
257    Ok(StoredEvent {
258        sequence: sequence_from_sql(sequence)?,
259        event_id: EventId::from(event_id),
260        account_id: account.clone(),
261        case_key: CaseKey::new(workflow_key, case_id),
262        case_revision: revision_from_sql(revision)?,
263        command_id: CommandId::from(command_id),
264        event_type: column(row, "event_type")?,
265        payload: column(row, "payload")?,
266        occurred_at: column(row, "occurred_at")?,
267        redaction: decode_redaction(row)?,
268    })
269}
270
271#[async_trait]
272impl EventJournalReader for PgStores {
273    async fn list_since(
274        &self,
275        account: &AccountId,
276        case_key: &CaseKey,
277        since: CaseRevision,
278        limit: usize,
279    ) -> Result<Vec<StoredEvent>, StoreError> {
280        let mut conn = self.connection().await?;
281        list_since(&mut conn, account, case_key, since, limit).await
282    }
283
284    async fn read_from(
285        &self,
286        account: &AccountId,
287        after: EventCursor,
288        limit: usize,
289    ) -> Result<EventPage, StoreError> {
290        let mut conn = self.connection().await?;
291        read_from(&mut conn, account, after, limit).await
292    }
293
294    async fn get_by_ids(
295        &self,
296        account: &AccountId,
297        ids: &[EventId],
298    ) -> Result<Vec<StoredEvent>, StoreError> {
299        let mut conn = self.connection().await?;
300        get_by_ids(&mut conn, account, ids).await
301    }
302
303    async fn count(&self, account: &AccountId, case_key: &CaseKey) -> Result<u64, StoreError> {
304        let mut conn = self.connection().await?;
305        count(&mut conn, account, case_key).await
306    }
307}
308
309#[async_trait]
310impl EventJournalWriter for PgStores {
311    async fn append(&self, batch: EventBatch) -> Result<Vec<EventId>, StoreError> {
312        let mut transaction = self.transaction().await?;
313        let ids = append(&mut transaction, batch).await?;
314        commit(transaction).await?;
315        Ok(ids)
316    }
317
318    async fn redact_payload(
319        &self,
320        account: &AccountId,
321        event_id: &EventId,
322        authority: &RedactionAuthority,
323    ) -> Result<EventRedaction, StoreError> {
324        let mut transaction = self.transaction().await?;
325        let record = redact_payload(&mut transaction, account, event_id, authority).await?;
326        commit(transaction).await?;
327        Ok(record)
328    }
329}
turnframe_store_postgres/events.rs

turnframe_store_postgres/
events.rs
