coven_database/
schema_introspection.rs

1use std::collections::BTreeMap;
2
3use rusqlite::{Connection, OptionalExtension};
4
5#[derive(Debug, thiserror::Error)]
6pub enum CreateTableSchemaError {
7    #[error("read CREATE TABLE schema for {table:?} failed: {source}")]
8    Read {
9        table: String,
10        #[source]
11        source: rusqlite::Error,
12    },
13    #[error("no CREATE TABLE schema for {0}")]
14    Missing(String),
15    #[error("bad CREATE TABLE SQL for {table}: {sql}")]
16    Malformed { table: String, sql: String },
17}
18
19/// `CREATE TABLE` text for `table` from `sqlite_master`.
20pub(crate) fn create_table_sql(
21    conn: &Connection,
22    table: &str,
23) -> Result<String, CreateTableSchemaError> {
24    let create = conn
25        .query_row(
26            "SELECT sql FROM sqlite_master WHERE type='table' AND name = ?1",
27            [table],
28            |row| row.get::<_, Option<String>>(0),
29        )
30        .optional()
31        .map_err(|source| CreateTableSchemaError::Read {
32            table: table.to_string(),
33            source,
34        })?;
35    create
36        .flatten()
37        .ok_or_else(|| CreateTableSchemaError::Missing(table.to_string()))
38}
39
40/// Compare schema tokens without interpreting formatting as a schema change.
41/// Quoted names and text remain byte-for-byte intact.
42pub(crate) fn normalize_schema_sql(sql: &str) -> rusqlite::Result<String> {
43    use sqlite3_parser::lexer::{sql::Tokenizer, Scanner};
44
45    let mut scanner = Scanner::new(Tokenizer::new());
46    let mut normalized = String::with_capacity(sql.len());
47    loop {
48        let (start, token, end) = scanner
49            .scan(sql.as_bytes())
50            .map_err(|error| rusqlite::Error::ToSqlConversionFailure(Box::new(error)))?;
51        if token.is_none() {
52            return Ok(normalized);
53        }
54        // Scanner values omit delimiters for some tokens (notably blobs), so
55        // use its source span to preserve each token's complete representation.
56        let text = &sql[start..end];
57        if !normalized.is_empty() {
58            normalized.push(' ');
59        }
60        if text.starts_with(['\'', '"', '`', '[']) {
61            normalized.push_str(text);
62        } else {
63            normalized.push_str(&text.to_ascii_lowercase());
64        }
65    }
66}
67
68/// Qualify a `CREATE TABLE <name> ...` statement so it builds the table inside
69/// the attached schema `alias`, replacing only the table-name token.
70pub fn rewrite_create_into_schema(
71    create: &str,
72    table: &str,
73    alias: &str,
74) -> Result<String, CreateTableSchemaError> {
75    let Some((name_start, name_end, parsed_table)) = create_table_name_token(create) else {
76        return Err(CreateTableSchemaError::Malformed {
77            table: table.to_string(),
78            sql: create.to_string(),
79        });
80    };
81    if parsed_table != table {
82        return Err(CreateTableSchemaError::Malformed {
83            table: table.to_string(),
84            sql: create.to_string(),
85        });
86    }
87
88    let qualified = format!("{alias}.{}", quote_ident(table));
89    let mut out = String::with_capacity(create.len() + qualified.len());
90    out.push_str(&create[..name_start]);
91    out.push_str(&qualified);
92    out.push_str(&create[name_end..]);
93    Ok(out)
94}
95
96fn create_table_name_token(create: &str) -> Option<(usize, usize, String)> {
97    let mut pos = consume_keyword_ws(create, skip_ascii_ws(create, 0), "CREATE")?;
98    pos = consume_keyword_ws(create, pos, "TABLE")?;
99
100    if keyword_at(create, pos, "IF") {
101        pos = consume_keyword_ws(create, pos, "IF")?;
102        pos = consume_keyword_ws(create, pos, "NOT")?;
103        pos = consume_keyword_ws(create, pos, "EXISTS")?;
104    }
105
106    parse_identifier_token(create, pos)
107}
108
109fn skip_ascii_ws(sql: &str, mut pos: usize) -> usize {
110    while sql.as_bytes().get(pos).is_some_and(u8::is_ascii_whitespace) {
111        pos += 1;
112    }
113    pos
114}
115
116fn keyword_at(sql: &str, pos: usize, keyword: &str) -> bool {
117    let Some(end) = pos.checked_add(keyword.len()) else {
118        return false;
119    };
120    sql.get(pos..end)
121        .is_some_and(|token| token.eq_ignore_ascii_case(keyword))
122        && sql.as_bytes().get(end).is_some_and(u8::is_ascii_whitespace)
123}
124
125fn consume_keyword(sql: &str, pos: usize, keyword: &str) -> Option<usize> {
126    keyword_at(sql, pos, keyword).then_some(pos + keyword.len())
127}
128
129fn consume_keyword_ws(sql: &str, pos: usize, keyword: &str) -> Option<usize> {
130    Some(skip_ascii_ws(sql, consume_keyword(sql, pos, keyword)?))
131}
132
133fn parse_identifier_token(sql: &str, pos: usize) -> Option<(usize, usize, String)> {
134    match sql.as_bytes().get(pos).copied()? {
135        b'"' => parse_delimited_identifier(sql, pos, b'"'),
136        _ => parse_bare_identifier(sql, pos),
137    }
138}
139
140fn parse_delimited_identifier(
141    sql: &str,
142    start: usize,
143    delimiter: u8,
144) -> Option<(usize, usize, String)> {
145    let bytes = sql.as_bytes();
146    let mut pos = start + 1;
147    let mut out = String::new();
148    while pos < bytes.len() {
149        if bytes[pos] == delimiter {
150            if bytes.get(pos + 1).copied() == Some(delimiter) {
151                out.push(delimiter as char);
152                pos += 2;
153            } else {
154                return Some((start, pos + 1, out));
155            }
156        } else {
157            let ch = sql[pos..].chars().next()?;
158            out.push(ch);
159            pos += ch.len_utf8();
160        }
161    }
162    None
163}
164
165fn parse_bare_identifier(sql: &str, start: usize) -> Option<(usize, usize, String)> {
166    let mut pos = start;
167    while pos < sql.len() {
168        let b = sql.as_bytes()[pos];
169        if b.is_ascii_whitespace() || b == b'(' {
170            break;
171        }
172        let ch = sql[pos..].chars().next()?;
173        pos += ch.len_utf8();
174    }
175    (pos > start).then(|| (start, pos, sql[start..pos].to_string()))
176}
177
178/// Column names of `table`, in declared order, via `PRAGMA table_info`. The
179/// index of a name here is the index SQLite session changesets report for that
180/// column.
181pub(crate) fn table_columns(conn: &Connection, table: &str) -> rusqlite::Result<Vec<String>> {
182    let sql = format!("PRAGMA table_info({})", quote_ident(table));
183    let mut stmt = conn.prepare(&sql)?;
184    let columns = stmt
185        .query_map([], |row| row.get::<_, String>(1))?
186        .collect::<Result<Vec<_>, _>>()?;
187    Ok(columns)
188}
189
190/// Quote an SQL identifier (table/column name), doubling any embedded quote, so
191/// a trusted-but-unbindable name interpolates safely. Identifiers cannot be
192/// passed as bound parameters; this is the safe interpolation path for them.
193pub fn quote_ident(ident: &str) -> String {
194    format!("\"{}\"", ident.replace('"', "\"\""))
195}
196
197#[derive(Debug, thiserror::Error)]
198pub enum ForeignKeySchemaError {
199    #[error(transparent)]
200    Sqlite(#[from] rusqlite::Error),
201    #[error("foreign key on {child_table:?} is malformed: {reason}")]
202    Malformed { child_table: String, reason: String },
203}
204
205#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord)]
206pub struct ForeignKeyColumn {
207    pub child: String,
208    pub parent: String,
209}
210
211#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord)]
212pub struct ForeignKeyEdge {
213    pub parent_table: String,
214    pub columns: Vec<ForeignKeyColumn>,
215    pub on_update: String,
216    pub on_delete: String,
217    pub match_clause: String,
218}
219
220struct ForeignKeyRow {
221    sequence: i64,
222    parent_table: String,
223    child_column: String,
224    parent_column: Option<String>,
225    on_update: String,
226    on_delete: String,
227    match_clause: String,
228}
229
230/// Every outgoing foreign key on `child_table`, grouped by constraint and
231/// sorted by its complete parent/column/action shape. SQLite's constraint ids
232/// and PRAGMA listing order are deliberately discarded.
233pub(crate) fn foreign_key_edges(
234    conn: &Connection,
235    child_table: &str,
236) -> Result<Vec<ForeignKeyEdge>, ForeignKeySchemaError> {
237    let sql = format!("PRAGMA foreign_key_list({})", quote_ident(child_table));
238    let mut statement = conn.prepare(&sql)?;
239    let rows = statement.query_map([], |row| {
240        Ok((
241            row.get::<_, i64>(0)?,
242            ForeignKeyRow {
243                sequence: row.get(1)?,
244                parent_table: row.get(2)?,
245                child_column: row.get(3)?,
246                parent_column: row.get(4)?,
247                on_update: row.get::<_, String>(5)?.to_ascii_uppercase(),
248                on_delete: row.get::<_, String>(6)?.to_ascii_uppercase(),
249                match_clause: row.get::<_, String>(7)?.to_ascii_uppercase(),
250            },
251        ))
252    })?;
253    let mut grouped: BTreeMap<i64, Vec<ForeignKeyRow>> = BTreeMap::new();
254    for row in rows {
255        let (id, row) = row?;
256        grouped.entry(id).or_default().push(row);
257    }
258
259    let mut edges = Vec::with_capacity(grouped.len());
260    for mut rows in grouped.into_values() {
261        rows.sort_by_key(|row| row.sequence);
262        let first = rows
263            .first()
264            .ok_or_else(|| ForeignKeySchemaError::Malformed {
265                child_table: child_table.to_string(),
266                reason: "constraint has no columns".to_string(),
267            })?;
268        if rows.iter().any(|row| {
269            row.parent_table != first.parent_table
270                || row.on_update != first.on_update
271                || row.on_delete != first.on_delete
272                || row.match_clause != first.match_clause
273        }) {
274            return Err(ForeignKeySchemaError::Malformed {
275                child_table: child_table.to_string(),
276                reason: "one constraint reports inconsistent parent or actions".to_string(),
277            });
278        }
279        let omitted_parent_columns = rows.iter().all(|row| row.parent_column.is_none());
280        if !omitted_parent_columns && rows.iter().any(|row| row.parent_column.is_none()) {
281            return Err(ForeignKeySchemaError::Malformed {
282                child_table: child_table.to_string(),
283                reason: "one constraint mixes named and omitted parent columns".to_string(),
284            });
285        }
286        let inferred_parent_columns = if omitted_parent_columns {
287            primary_key_columns(conn, &first.parent_table)?
288        } else {
289            Vec::new()
290        };
291        if omitted_parent_columns && inferred_parent_columns.len() != rows.len() {
292            return Err(ForeignKeySchemaError::Malformed {
293                child_table: child_table.to_string(),
294                reason: format!(
295                    "{} child columns reference {} primary-key columns",
296                    rows.len(),
297                    inferred_parent_columns.len(),
298                ),
299            });
300        }
301        let columns = rows
302            .iter()
303            .enumerate()
304            .map(|(position, row)| ForeignKeyColumn {
305                child: row.child_column.clone(),
306                parent: row
307                    .parent_column
308                    .clone()
309                    .unwrap_or_else(|| inferred_parent_columns[position].clone()),
310            })
311            .collect();
312        edges.push(ForeignKeyEdge {
313            parent_table: first.parent_table.clone(),
314            columns,
315            on_update: first.on_update.clone(),
316            on_delete: first.on_delete.clone(),
317            match_clause: first.match_clause.clone(),
318        });
319    }
320    edges.sort();
321    Ok(edges)
322}
323
324fn primary_key_columns(
325    conn: &Connection,
326    table: &str,
327) -> Result<Vec<String>, ForeignKeySchemaError> {
328    let sql = format!("PRAGMA table_info({})", quote_ident(table));
329    let mut statement = conn.prepare(&sql)?;
330    let rows = statement.query_map([], |row| {
331        Ok((row.get::<_, i64>(5)?, row.get::<_, String>(1)?))
332    })?;
333    let mut columns = rows
334        .collect::<rusqlite::Result<Vec<_>>>()?
335        .into_iter()
336        .filter(|(rank, _)| *rank > 0)
337        .collect::<Vec<_>>();
338    columns.sort_by_key(|(rank, _)| *rank);
339    Ok(columns.into_iter().map(|(_, name)| name).collect())
340}
341
342#[cfg(test)]
343mod tests {
344    use super::*;
345
346    #[test]
347    fn schema_normalization_preserves_unicode_literals_and_token_boundaries() {
348        let sql = "CREATE TABLE t ([é ] TEXT DEFAULT 'A  B''C', n BLOB DEFAULT X'4142') STRICT";
349        let normalized = normalize_schema_sql(sql).expect("normalize schema");
350        assert!(normalized.contains("[é ]"));
351        assert!(normalized.contains("'A  B''C'"));
352        assert!(normalized.contains("x'4142'"));
353        assert_eq!(
354            normalize_schema_sql("CREATE/*comment*/TABLE t(x TEXT) STRICT").expect("normalize"),
355            normalize_schema_sql("CREATE TABLE t ( x TEXT ) STRICT").expect("normalize"),
356        );
357    }
358
359    #[test]
360    fn schema_normalization_preserves_spaces_inside_bracketed_identifiers() {
361        assert_ne!(
362            normalize_schema_sql("CREATE TABLE t ([value ] TEXT) STRICT").expect("normalize"),
363            normalize_schema_sql("CREATE TABLE t ([value] TEXT) STRICT").expect("normalize"),
364        );
365    }
366
367    #[test]
368    fn create_table_rewrite_qualifies_table_token() {
369        let cases = [
370            (
371                "CREATE TABLE nodes (id TEXT PRIMARY KEY)",
372                "CREATE TABLE empty.\"nodes\" (id TEXT PRIMARY KEY)",
373            ),
374            (
375                "CREATE TABLE \"nodes\" (id TEXT PRIMARY KEY)",
376                "CREATE TABLE empty.\"nodes\" (id TEXT PRIMARY KEY)",
377            ),
378            (
379                "CREATE TABLE IF NOT EXISTS nodes (id TEXT PRIMARY KEY)",
380                "CREATE TABLE IF NOT EXISTS empty.\"nodes\" (id TEXT PRIMARY KEY)",
381            ),
382            (
383                "CREATE TABLE nodes (id TEXT PRIMARY KEY, parent_id TEXT REFERENCES \"nodes\" (id))",
384                "CREATE TABLE empty.\"nodes\" (id TEXT PRIMARY KEY, parent_id TEXT REFERENCES \"nodes\" (id))",
385            ),
386        ];
387
388        for (create, expected) in cases {
389            let rewritten = rewrite_create_into_schema(create, "nodes", "empty").expect("rewrite");
390            assert_eq!(rewritten, expected);
391        }
392    }
393
394    #[test]
395    fn create_table_rewrite_rejects_mismatched_table_token() {
396        let err = rewrite_create_into_schema(
397            "CREATE TABLE other_nodes (id TEXT PRIMARY KEY)",
398            "nodes",
399            "empty",
400        )
401        .expect_err("mismatched table token must fail");
402        assert!(
403            matches!(
404                err,
405                CreateTableSchemaError::Malformed { ref table, ref sql }
406                    if table == "nodes"
407                        && sql == "CREATE TABLE other_nodes (id TEXT PRIMARY KEY)"
408            ),
409            "unexpected error: {err}"
410        );
411    }
412}
coven_database/schema_introspection.rs

coven_database/
schema_introspection.rs
