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 {
17    Absent,
18    Present(Option<String>),
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.clone(),
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(
97    item: &rusqlite::session::ChangesetItem,
98    col: usize,
99    op: ChangeOp,
100    update_value: UpdateValue,
101) -> Result<(ColumnCell, 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 = matches!(new, ColumnCell::Present(_));
109            let cell = match update_value {
110                UpdateValue::New => match new {
111                    ColumnCell::Absent => old,
112                    present => present,
113                },
114                UpdateValue::Old => match old {
115                    ColumnCell::Absent => new,
116                    present => present,
117                },
118            };
119            Ok((cell, changed))
120        }
121    }
122}
123
124fn changeset_value(
125    item: &rusqlite::session::ChangesetItem,
126    col: usize,
127    side: UpdateValue,
128) -> Result<ColumnCell, ChangesetError> {
129    let value = match side {
130        UpdateValue::New => item.new_value(col),
131        UpdateValue::Old => item.old_value(col),
132    };
133    match value {
134        Ok(value) => Ok(ColumnCell::Present(value_ref_to_string(value))),
135        Err(rusqlite::Error::InvalidColumnIndex(_)) => Ok(ColumnCell::Absent),
136        Err(source) => Err(ChangesetError::Value {
137            side: match side {
138                UpdateValue::New => "new",
139                UpdateValue::Old => "old",
140            },
141            column: col,
142            source,
143        }),
144    }
145}
146
147/// Render a changeset/column [`ValueRef`] as an owned `String`, or `None` for
148/// SQL NULL. Mirrors `sqlite3_value_text`: text and blob bytes become a string
149/// (lossy on invalid UTF-8), and integers/reals their decimal text — so the
150/// `_updated_at` row-arbitration comparison and blob-plan column reads see the same strings the
151/// raw FFI path (gate.rs) produces.
152///
153/// Synced columns coven reads through here — `_updated_at`, gate columns, FK and
154/// blob-plan columns — are expected to be TEXT, INTEGER, or BLOB, never REAL. The
155/// REAL arm exists only so a stray float doesn't silently become `None`; it
156/// renders a faithful (round-tripping) decimal that always shows it is a float
157/// (a trailing `.0` when there is no fractional part or exponent), matching
158/// SQLite's REAL→text on whole numbers and simple decimals rather than diverging
159/// into Rust's integer-looking `f64::to_string` (`1.0` → `"1"`). It does not
160/// reproduce SQLite's exact scientific-notation threshold or 17th-digit rounding
161/// — there is no live impact, since no synced column is REAL.
162pub fn value_ref_to_string(v: ValueRef<'_>) -> Option<String> {
163    match v {
164        ValueRef::Null => None,
165        ValueRef::Integer(i) => Some(i.to_string()),
166        ValueRef::Real(f) => Some(real_to_sqlite_text(f)),
167        ValueRef::Text(t) | ValueRef::Blob(t) => Some(String::from_utf8_lossy(t).into_owned()),
168    }
169}
170
171/// Render a finite `f64` as a faithful decimal that always reads as a float, the
172/// way SQLite's REAL→text does for the common cases: a whole number keeps a
173/// trailing `.0` (`1.0` → `"1.0"`, not Rust's `"1"`), everything else is the
174/// shortest round-tripping decimal. Non-finite values render as SQLite spells
175/// them (`Inf`/`-Inf`); NaN cannot reach a well-formed synced column and renders
176/// empty rather than panicking.
177fn real_to_sqlite_text(f: f64) -> String {
178    if f.is_nan() {
179        return String::new();
180    }
181    if f.is_infinite() {
182        return if f < 0.0 {
183            "-Inf".to_string()
184        } else {
185            "Inf".to_string()
186        };
187    }
188    let s = f.to_string();
189    // Rust's shortest round-trip prints whole numbers as integers (`1`, `100`);
190    // SQLite always marks a float, so append `.0` when there is neither a decimal
191    // point nor an exponent.
192    if s.contains('.') || s.contains('e') || s.contains('E') {
193        s
194    } else {
195        format!("{s}.0")
196    }
197}
198
199#[cfg(test)]
200mod tests {
201    use super::*;
202
203    /// The REAL arm must render a faithful float: it round-trips back to the same
204    /// `f64`, always reads as a float (has a `.` or exponent), and matches SQLite's
205    /// `CAST(real AS TEXT)` on whole numbers and simple decimals — the cases that
206    /// would otherwise diverge via Rust's integer-looking `f64::to_string`. SQLite
207    /// is the ground truth the gate's raw-FFI `sqlite3_value_text` path uses.
208    #[test]
209    fn real_renders_as_a_faithful_float() {
210        let conn = rusqlite::Connection::open_in_memory().expect("open");
211        // Cases SQLite renders identically to our shortest-round-trip-plus-`.0`.
212        for &f in &[0.0_f64, 1.0, 1.5, -2.25, 0.1, 123456.789, 1.0e6, 100.0, 0.5] {
213            let sqlite_text: String = conn
214                .query_row("SELECT CAST(? AS TEXT)", [f], |r| r.get(0))
215                .expect("cast");
216            let ours = real_to_sqlite_text(f);
217            assert_eq!(
218                ours, sqlite_text,
219                "REAL {f} rendered {ours:?}, SQLite renders {sqlite_text:?}",
220            );
221        }
222
223        // The general invariant: round-trips and reads as a float, even where the
224        // exact spelling differs from SQLite (scientific threshold, 17th digit).
225        for &f in &[1.234567890123457_f64, 1.0e-7, 9_999_999_999_999.0, -42.0] {
226            let ours = real_to_sqlite_text(f);
227            assert!(
228                ours.contains('.') || ours.contains('e') || ours.contains('E'),
229                "REAL {f} rendered {ours:?} which doesn't read as a float",
230            );
231            assert_eq!(
232                ours.parse::<f64>().expect("parses back"),
233                f,
234                "REAL {f} rendered {ours:?} which doesn't round-trip",
235            );
236        }
237    }
238}
coven_database/changeset.rs

coven_database/
changeset.rs
