force_sync/store/pg/
journal.rs

1//! Journal write helpers for the `PostgreSQL` sync store.
2
3use chrono::{DateTime, Utc};
4use serde_json::Value;
5use tokio_postgres::GenericClient;
6
7use crate::{
8    error::ForceSyncError,
9    model::{ChangeEnvelope, ChangeOperation},
10};
11
12use super::PgStore;
13
14/// Result of attempting to append a journal entry when duplicates are allowed.
15#[derive(Debug, Clone, PartialEq, Eq)]
16pub enum AppendResult {
17    /// The row was inserted and the returned ID is the new journal row.
18    Inserted {
19        /// The inserted journal row ID.
20        journal_id: i64,
21    },
22    /// A row with the same `(source, source_cursor)` already exists.
23    Duplicate,
24}
25
26/// ⚡ Bolt: Values derived from a change envelope to be inserted into the journal.
27/// Lifetimes are used to borrow strings and the JSON payload directly to avoid unnecessary heap allocations per journal row.
28struct JournalValues<'a> {
29    tenant: &'a str,
30    object_name: &'a str,
31    external_id: &'a str,
32    source: &'static str,
33    source_cursor: String,
34    observed_at: DateTime<Utc>,
35    operation: &'static str,
36    tombstone: bool,
37    payload: &'a Value,
38    payload_hash: [u8; 32],
39}
40
41fn journal_values(envelope: &ChangeEnvelope) -> Result<JournalValues<'_>, ForceSyncError> {
42    let cursor = envelope
43        .cursor()
44        .ok_or(ForceSyncError::MissingSourceCursor)?;
45
46    Ok(JournalValues {
47        tenant: envelope.sync_key().tenant(),
48        object_name: envelope.sync_key().object_name(),
49        external_id: envelope.sync_key().external_id(),
50        source: envelope.source().as_db_value(),
51        source_cursor: cursor.as_db_value(),
52        observed_at: envelope.observed_at(),
53        operation: envelope.operation().as_db_value(),
54        tombstone: matches!(envelope.operation(), ChangeOperation::Delete),
55        payload: envelope.payload(),
56        payload_hash: envelope.payload_hash(),
57    })
58}
59
60async fn insert_journal<C>(client: &C, values: &JournalValues<'_>) -> Result<i64, ForceSyncError>
61where
62    C: GenericClient + Sync + ?Sized,
63{
64    let payload_hash = values.payload_hash.as_slice();
65    let row = client
66        .query_one(
67            "insert into sync_journal (
68                tenant,
69                object_name,
70                external_id,
71                source,
72                source_cursor,
73                observed_at,
74                operation,
75                tombstone,
76                payload,
77                payload_hash,
78                schema_version
79            ) values (
80                $1,
81                $2,
82                $3,
83                $4,
84                $5,
85                $6::timestamptz,
86                $7,
87                $8,
88                $9::jsonb,
89                $10,
90                1
91            ) returning journal_id",
92            &[
93                &values.tenant,
94                &values.object_name,
95                &values.external_id,
96                &values.source,
97                &values.source_cursor,
98                &values.observed_at,
99                &values.operation,
100                &values.tombstone,
101                &values.payload,
102                &payload_hash,
103            ],
104        )
105        .await?;
106
107    Ok(row.get(0))
108}
109
110async fn insert_journal_if_new<C>(
111    client: &C,
112    values: &JournalValues<'_>,
113) -> Result<Option<i64>, ForceSyncError>
114where
115    C: GenericClient + Sync + ?Sized,
116{
117    let payload_hash = values.payload_hash.as_slice();
118    let row = client
119        .query_opt(
120            "insert into sync_journal (
121                tenant,
122                object_name,
123                external_id,
124                source,
125                source_cursor,
126                observed_at,
127                operation,
128                tombstone,
129                payload,
130                payload_hash,
131                schema_version
132            ) values (
133                $1,
134                $2,
135                $3,
136                $4,
137                $5,
138                $6::timestamptz,
139                $7,
140                $8,
141                $9::jsonb,
142                $10,
143                1
144            ) on conflict (source, source_cursor) do nothing
145            returning journal_id",
146            &[
147                &values.tenant,
148                &values.object_name,
149                &values.external_id,
150                &values.source,
151                &values.source_cursor,
152                &values.observed_at,
153                &values.operation,
154                &values.tombstone,
155                &values.payload,
156                &payload_hash,
157            ],
158        )
159        .await?;
160
161    Ok(row.map(|row| row.get(0)))
162}
163
164impl PgStore {
165    /// Appends a journal entry and returns its database identifier.
166    ///
167    /// # Errors
168    ///
169    /// Returns an error if the cursor is missing or the database write fails.
170    pub async fn append_journal(&self, envelope: &ChangeEnvelope) -> Result<i64, ForceSyncError> {
171        let values = journal_values(envelope)?;
172        let client = self.pool().get().await?;
173        insert_journal(&**client, &values).await
174    }
175
176    /// Appends a journal entry if the `(source, source_cursor)` pair is new.
177    ///
178    /// # Errors
179    ///
180    /// Returns an error if the cursor is missing or the database write fails.
181    pub async fn append_journal_if_new(
182        &self,
183        envelope: &ChangeEnvelope,
184    ) -> Result<AppendResult, ForceSyncError> {
185        let values = journal_values(envelope)?;
186        let client = self.pool().get().await?;
187        insert_journal_if_new(&**client, &values)
188            .await?
189            .map_or_else(
190                || Ok(AppendResult::Duplicate),
191                |journal_id| Ok(AppendResult::Inserted { journal_id }),
192            )
193    }
194
195    /// Appends a journal row in an existing transaction.
196    ///
197    /// This helper exists so same-transaction workflows can remain atomic
198    /// without forcing the caller through another store method.
199    ///
200    /// # Errors
201    ///
202    /// Returns an error if the cursor is missing or the database write fails.
203    pub async fn append_journal_in_tx<C>(
204        client: &C,
205        envelope: &ChangeEnvelope,
206    ) -> Result<i64, ForceSyncError>
207    where
208        C: GenericClient + Sync + ?Sized,
209    {
210        let values = journal_values(envelope)?;
211        insert_journal(client, &values).await
212    }
213
214    /// Appends a journal row in an existing transaction, skipping duplicates.
215    ///
216    /// # Errors
217    ///
218    /// Returns an error if the cursor is missing or the database write fails.
219    pub async fn append_journal_if_new_in_tx<C>(
220        client: &C,
221        envelope: &ChangeEnvelope,
222    ) -> Result<AppendResult, ForceSyncError>
223    where
224        C: GenericClient + Sync + ?Sized,
225    {
226        let values = journal_values(envelope)?;
227
228        insert_journal_if_new(client, &values).await?.map_or_else(
229            || Ok(AppendResult::Duplicate),
230            |journal_id| Ok(AppendResult::Inserted { journal_id }),
231        )
232    }
233}
234
235#[cfg(test)]
236mod tests {
237    use chrono::Utc;
238    use serde_json::json;
239
240    use crate::{
241        identity::SyncKey,
242        model::{ChangeEnvelope, ChangeOperation, SourceCursor, SourceSystem},
243    };
244
245    use super::journal_values;
246
247    #[test]
248    fn journal_values_keep_observed_at_as_datetime() {
249        let observed_at = Utc::now();
250        let envelope = ChangeEnvelope::new(
251            SyncKey::new("tenant", "Account", "external-1")
252                .unwrap_or_else(|error| panic!("unexpected sync key construction error: {error}")),
253            SourceSystem::Postgres,
254            ChangeOperation::Upsert,
255            observed_at,
256            json!({"Name": "Acme"}),
257        )
258        .with_cursor(SourceCursor::PostgresLsn("lsn-1".to_owned()));
259
260        let values = journal_values(&envelope)
261            .unwrap_or_else(|error| panic!("unexpected journal values error: {error}"));
262
263        let _: chrono::DateTime<chrono::Utc> = values.observed_at;
264        let _: &serde_json::Value = values.payload;
265        assert_eq!(values.observed_at, observed_at);
266        assert_eq!(*values.payload, json!({"Name": "Acme"}));
267    }
268}
force_sync/store/pg/journal.rs

force_sync/store/pg/
journal.rs
