coven_database/
changeset.rs

1//! Decode SQLite session changesets into database-independent row changes.
2
3use fallible_streaming_iterator::FallibleStreamingIterator;
4use rusqlite::hooks::Action;
5use rusqlite::session::ChangesetIter;
6use rusqlite::types::ValueRef;
7
8use coven_foundation::changeset::{ChangeOp, RowChange};
9
10#[derive(Debug, Clone, Copy, PartialEq, Eq)]
11pub(crate) enum UpdateValue {
12    New,
13    Old,
14}
15
16enum ColumnCell<'value> {
17    Absent,
18    Present(ValueRef<'value>),
19}
20
21#[derive(Debug, thiserror::Error)]
22pub enum ChangesetError {
23    #[error("start changeset iterator: {0}")]
24    Start(#[source] rusqlite::Error),
25    #[error("advance changeset iterator: {0}")]
26    Next(#[source] rusqlite::Error),
27    #[error("read changeset operation: {0}")]
28    Operation(#[source] rusqlite::Error),
29    #[error("read changeset {side:?} value for column {column}: {source}")]
30    Value {
31        side: &'static str,
32        column: usize,
33        #[source]
34        source: rusqlite::Error,
35    },
36}
37
38/// Walk a changeset and return every row change with its column values.
39///
40/// Returns an empty vec for an empty changeset.
41pub fn walk(changeset_bytes: &[u8]) -> Result<Vec<RowChange>, ChangesetError> {
42    walk_with_update_values(changeset_bytes, UpdateValue::New)
43}
44
45pub fn walk_old(changeset_bytes: &[u8]) -> Result<Vec<RowChange>, ChangesetError> {
46    walk_with_update_values(changeset_bytes, UpdateValue::Old)
47}
48
49fn walk_with_update_values(
50    changeset_bytes: &[u8],
51    update_value: UpdateValue,
52) -> Result<Vec<RowChange>, ChangesetError> {
53    if changeset_bytes.is_empty() {
54        return Ok(Vec::new());
55    }
56
57    let input: &mut dyn std::io::Read = &mut &changeset_bytes[..];
58    let mut iter = ChangesetIter::start_strm(&input).map_err(ChangesetError::Start)?;
59
60    let mut changes = Vec::new();
61    while let Some(item) = iter.next().map_err(ChangesetError::Next)? {
62        let op = item.op().map_err(ChangesetError::Operation)?;
63        let change_op = match op.code() {
64            Action::SQLITE_INSERT => ChangeOp::Insert,
65            Action::SQLITE_UPDATE => ChangeOp::Update,
66            Action::SQLITE_DELETE => ChangeOp::Delete,
67            _ => continue,
68        };
69        let ncol = op.number_of_columns();
70
71        let cells = (0..ncol)
72            .map(|c| extract_col(item, c as usize, change_op, update_value))
73            .collect::<Result<Vec<_>, _>>()?;
74        let columns = cells
75            .iter()
76            .map(|(cell, _)| match cell {
77                ColumnCell::Absent => None,
78                ColumnCell::Present(value) => value_ref_to_string(*value),
79            })
80            .collect();
81        let changed_columns = cells.iter().map(|(_, changed)| *changed).collect();
82        changes.push(RowChange::new(
83            op.table_name().to_string(),
84            change_op,
85            columns,
86            changed_columns,
87        ));
88    }
89
90    Ok(changes)
91}
92
93/// Extract a column value from a changeset item following the op's old/new
94/// semantics. An absent column (unchanged in an update) reads as an
95/// `InvalidColumnIndex` error from rusqlite, which maps to `None`.
96fn extract_col<'value>(
97    item: &'value rusqlite::session::ChangesetItem,
98    col: usize,
99    op: ChangeOp,
100    update_value: UpdateValue,
101) -> Result<(ColumnCell<'value>, bool), ChangesetError> {
102    match op {
103        ChangeOp::Insert => changeset_value(item, col, UpdateValue::New).map(|cell| (cell, true)),
104        ChangeOp::Delete => changeset_value(item, col, UpdateValue::Old).map(|cell| (cell, true)),
105        ChangeOp::Update => {
106            let new = changeset_value(item, col, UpdateValue::New)?;
107            let old = changeset_value(item, col, UpdateValue::Old)?;
108            let changed = match (&new, &old) {
109                (ColumnCell::Present(new), ColumnCell::Present(old)) => new != old,
110                (ColumnCell::Present(_), ColumnCell::Absent) => true,
111                (ColumnCell::Absent, _) => false,
112            };
113            let cell = match update_value {
114                UpdateValue::New => match new {
115                    ColumnCell::Absent => old,
116                    present => present,
117                },
118                UpdateValue::Old => match old {
119                    ColumnCell::Absent => new,
120                    present => present,
121                },
122            };
123            Ok((cell, changed))
124        }
125    }
126}
127
128fn changeset_value<'value>(
129    item: &'value rusqlite::session::ChangesetItem,
130    col: usize,
131    side: UpdateValue,
132) -> Result<ColumnCell<'value>, ChangesetError> {
133    let value = match side {
134        UpdateValue::New => item.new_value(col),
135        UpdateValue::Old => item.old_value(col),
136    };
137    match value {
138        Ok(value) => Ok(ColumnCell::Present(value)),
139        Err(rusqlite::Error::InvalidColumnIndex(_)) => Ok(ColumnCell::Absent),
140        Err(source) => Err(ChangesetError::Value {
141            side: match side {
142                UpdateValue::New => "new",
143                UpdateValue::Old => "old",
144            },
145            column: col,
146            source,
147        }),
148    }
149}
150
151/// Render a changeset/column [`ValueRef`] as an owned `String`, or `None` for
152/// SQL NULL. Mirrors `sqlite3_value_text`: text and blob bytes become a string
153/// (lossy on invalid UTF-8), and integers/reals their decimal text — so the
154/// `_updated_at` row-arbitration comparison and blob-plan column reads see the same strings the
155/// raw FFI path (gate.rs) produces.
156///
157/// Synced columns coven reads through here — `_updated_at`, gate columns, FK and
158/// blob-plan columns — are expected to be TEXT, INTEGER, or BLOB, never REAL. The
159/// REAL arm exists only so a stray float doesn't silently become `None`; it
160/// renders a faithful (round-tripping) decimal that always shows it is a float
161/// (a trailing `.0` when there is no fractional part or exponent), matching
162/// SQLite's REAL→text on whole numbers and simple decimals rather than diverging
163/// into Rust's integer-looking `f64::to_string` (`1.0` → `"1"`). It does not
164/// reproduce SQLite's exact scientific-notation threshold or 17th-digit rounding
165/// — there is no live impact, since no synced column is REAL.
166pub fn value_ref_to_string(v: ValueRef<'_>) -> Option<String> {
167    match v {
168        ValueRef::Null => None,
169        ValueRef::Integer(i) => Some(i.to_string()),
170        ValueRef::Real(f) => Some(real_to_sqlite_text(f)),
171        ValueRef::Text(t) | ValueRef::Blob(t) => Some(String::from_utf8_lossy(t).into_owned()),
172    }
173}
174
175/// Render a finite `f64` as a faithful decimal that always reads as a float, the
176/// way SQLite's REAL→text does for the common cases: a whole number keeps a
177/// trailing `.0` (`1.0` → `"1.0"`, not Rust's `"1"`), everything else is the
178/// shortest round-tripping decimal. Non-finite values render as SQLite spells
179/// them (`Inf`/`-Inf`); NaN cannot reach a well-formed synced column and renders
180/// empty rather than panicking.
181fn real_to_sqlite_text(f: f64) -> String {
182    if f.is_nan() {
183        return String::new();
184    }
185    if f.is_infinite() {
186        return if f < 0.0 {
187            "-Inf".to_string()
188        } else {
189            "Inf".to_string()
190        };
191    }
192    let s = f.to_string();
193    // Rust's shortest round-trip prints whole numbers as integers (`1`, `100`);
194    // SQLite always marks a float, so append `.0` when there is neither a decimal
195    // point nor an exponent.
196    if s.contains('.') || s.contains('e') || s.contains('E') {
197        s
198    } else {
199        format!("{s}.0")
200    }
201}
202
203#[cfg(test)]
204mod tests {
205    use super::*;
206
207    /// The REAL arm must render a faithful float: it round-trips back to the same
208    /// `f64`, always reads as a float (has a `.` or exponent), and matches SQLite's
209    /// `CAST(real AS TEXT)` on whole numbers and simple decimals — the cases that
210    /// would otherwise diverge via Rust's integer-looking `f64::to_string`. SQLite
211    /// is the ground truth the gate's raw-FFI `sqlite3_value_text` path uses.
212    #[test]
213    fn real_renders_as_a_faithful_float() {
214        let conn = rusqlite::Connection::open_in_memory().expect("open");
215        // Cases SQLite renders identically to our shortest-round-trip-plus-`.0`.
216        for &f in &[0.0_f64, 1.0, 1.5, -2.25, 0.1, 123456.789, 1.0e6, 100.0, 0.5] {
217            let sqlite_text: String = conn
218                .query_row("SELECT CAST(? AS TEXT)", [f], |r| r.get(0))
219                .expect("cast");
220            let ours = real_to_sqlite_text(f);
221            assert_eq!(
222                ours, sqlite_text,
223                "REAL {f} rendered {ours:?}, SQLite renders {sqlite_text:?}",
224            );
225        }
226
227        // The general invariant: round-trips and reads as a float, even where the
228        // exact spelling differs from SQLite (scientific threshold, 17th digit).
229        for &f in &[1.234567890123457_f64, 1.0e-7, 9_999_999_999_999.0, -42.0] {
230            let ours = real_to_sqlite_text(f);
231            assert!(
232                ours.contains('.') || ours.contains('e') || ours.contains('E'),
233                "REAL {f} rendered {ours:?} which doesn't read as a float",
234            );
235            assert_eq!(
236                ours.parse::<f64>().expect("parses back"),
237                f,
238                "REAL {f} rendered {ours:?} which doesn't round-trip",
239            );
240        }
241    }
242}
coven_database/changeset.rs

coven_database/
changeset.rs
