coven_core/
changeset.rs

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

coven_core/
changeset.rs
