sim_platform_sqlite/driver/
introspection.rs

1use super::{
2    AdoptionManifest, BaseDomain, ColumnName, Connection, IndexName, PhysicalColumn, PhysicalIndex,
3    PhysicalSchema, PhysicalTable, ProviderName, RevisionName, SchemaAttestation, SchemaName,
4    SiteError, Symbol, TableName, map_error, provider_symbol, relation_id_text, valid_source,
5};
6
7/// Normalizes the main `SQLite` catalog, excluding capsule metadata.
8///
9/// # Errors
10///
11/// Returns a typed site refusal when catalog queries fail or observed names,
12/// domains, columns, indexes, or revisions cannot be normalized.
13pub fn introspect_connection(
14    connection: &Connection,
15    revision: RevisionName,
16) -> Result<PhysicalSchema, SiteError> {
17    let mut tables_stmt = connection.prepare("SELECT name FROM sqlite_schema WHERE type='table' AND name NOT LIKE 'sqlite_%' AND name NOT LIKE '__sim_%' ORDER BY name").map_err(|error| map_error(&error))?;
18    let names = tables_stmt
19        .query_map([], |row| row.get::<_, String>(0))
20        .map_err(|error| map_error(&error))?
21        .collect::<Result<Vec<_>, _>>()
22        .map_err(|error| map_error(&error))?;
23    let mut tables = Vec::new();
24    for name in names {
25        if !valid_source(&Symbol::new(name.as_str())) {
26            return Err(SiteError::Conversion);
27        }
28        let mut columns_stmt = connection
29            .prepare(&format!(
30                "PRAGMA table_info(\"{}\")",
31                name.replace('"', "\"\"")
32            ))
33            .map_err(|error| map_error(&error))?;
34        let columns = columns_stmt
35            .query_map([], |row| {
36                Ok((
37                    row.get::<_, i64>(0)?,
38                    row.get::<_, String>(1)?,
39                    row.get::<_, String>(2)?,
40                    row.get::<_, i64>(3)?,
41                ))
42            })
43            .map_err(|error| map_error(&error))?
44            .map(|result| {
45                let (ordinal, name, ty, notnull) = result.map_err(|error| map_error(&error))?;
46                let (domain, storage) = affinity(&ty);
47                Ok(PhysicalColumn {
48                    name: ColumnName::new(Symbol::new(name)).map_err(|_| SiteError::Conversion)?,
49                    domain: domain.id(),
50                    storage,
51                    nullable: notnull == 0,
52                    ordinal: u32::try_from(ordinal).map_err(|_| SiteError::Conversion)?,
53                })
54            })
55            .collect::<Result<Vec<_>, SiteError>>()?;
56        let mut indexes_stmt = connection
57            .prepare(&format!(
58                "PRAGMA index_list(\"{}\")",
59                name.replace('"', "\"\"")
60            ))
61            .map_err(|error| map_error(&error))?;
62        let index_rows = indexes_stmt
63            .query_map([], |row| {
64                Ok((row.get::<_, String>(1)?, row.get::<_, i64>(2)?))
65            })
66            .map_err(|error| map_error(&error))?
67            .collect::<Result<Vec<_>, _>>()
68            .map_err(|error| map_error(&error))?;
69        let mut indexes = Vec::new();
70        for (index_name, unique) in index_rows {
71            if index_name.starts_with("sqlite_") {
72                continue;
73            }
74            let mut info = connection
75                .prepare(&format!(
76                    "PRAGMA index_info(\"{}\")",
77                    index_name.replace('"', "\"\"")
78                ))
79                .map_err(|error| map_error(&error))?;
80            let keys = info
81                .query_map([], |row| row.get::<_, String>(2))
82                .map_err(|error| map_error(&error))?
83                .map(|v| {
84                    ColumnName::new(Symbol::new(v.map_err(|error| map_error(&error))?))
85                        .map_err(|_| SiteError::Conversion)
86                })
87                .collect::<Result<Vec<_>, _>>()?;
88            indexes.push(PhysicalIndex {
89                name: IndexName::new(Symbol::new(index_name)).map_err(|_| SiteError::Conversion)?,
90                columns: keys,
91                unique: unique != 0,
92            });
93        }
94        tables.push(PhysicalTable {
95            name: TableName::new(Symbol::new(name)).map_err(|_| SiteError::Conversion)?,
96            columns,
97            indexes,
98        });
99    }
100    PhysicalSchema::normalize(
101        ProviderName::new(provider_symbol()).map_err(|_| SiteError::Conversion)?,
102        SchemaName::new(Symbol::new("main")).map_err(|_| SiteError::Conversion)?,
103        revision,
104        tables,
105    )
106    .map_err(|_| SiteError::Conversion)
107}
108
109/// Verifies existing capsule metadata or atomically adopts an exact old file.
110///
111/// Adoption cannot bless drift: the authored physical identity must equal a
112/// fresh normalized catalog observation before metadata is written.
113///
114/// # Errors
115///
116/// Returns a typed site refusal when introspection fails, retained attestation
117/// metadata disagrees, adoption proof is absent or invalid, or the metadata
118/// transaction cannot commit atomically.
119pub fn verify_or_adopt(
120    connection: &mut Connection,
121    logical_schema: sim_relation_core::RelationId,
122    revision: sim_relation_core::RelationId,
123    revision_name: RevisionName,
124    adoption: Option<&AdoptionManifest>,
125) -> Result<SchemaAttestation, SiteError> {
126    let physical_schema = introspect_connection(connection, revision_name)?
127        .id()
128        .map_err(|_| SiteError::Provider)?;
129    let existing = connection
130        .query_row(
131            "SELECT logical_schema, physical_schema, revision FROM __sim_relation_attestation WHERE singleton=1",
132            [],
133            |row| Ok((row.get::<_, String>(0)?, row.get::<_, String>(1)?, row.get::<_, String>(2)?)),
134        )
135        .ok();
136    if let Some((logical, physical, recorded_revision)) = existing {
137        if logical != relation_id_text(&logical_schema)
138            || physical != relation_id_text(&physical_schema)
139            || recorded_revision != relation_id_text(&revision)
140        {
141            return Err(SiteError::Drift);
142        }
143    } else {
144        adoption
145            .ok_or(SiteError::Drift)?
146            .verify(&physical_schema)
147            .map_err(|_| SiteError::Drift)?;
148        let transaction = connection
149            .transaction()
150            .map_err(|error| map_error(&error))?;
151        transaction.execute_batch("CREATE TABLE __sim_relation_attestation (singleton INTEGER PRIMARY KEY CHECK(singleton=1), logical_schema TEXT NOT NULL, physical_schema TEXT NOT NULL, revision TEXT NOT NULL)").map_err(|error| map_error(&error))?;
152        transaction
153            .execute(
154                "INSERT INTO __sim_relation_attestation VALUES (1, ?1, ?2, ?3)",
155                (
156                    relation_id_text(&logical_schema),
157                    relation_id_text(&physical_schema),
158                    relation_id_text(&revision),
159                ),
160            )
161            .map_err(|error| map_error(&error))?;
162        transaction.commit().map_err(|error| map_error(&error))?;
163    }
164    Ok(SchemaAttestation {
165        logical_schema,
166        physical_schema,
167        revision,
168    })
169}
170fn affinity(value: &str) -> (BaseDomain, sim_relation_core::StorageRepr) {
171    let upper = value.to_ascii_uppercase();
172    if upper.contains("INT") {
173        (BaseDomain::I64, sim_relation_core::StorageRepr::I64)
174    } else if upper.contains("CHAR") || upper.contains("CLOB") || upper.contains("TEXT") {
175        (BaseDomain::Text, sim_relation_core::StorageRepr::Text)
176    } else if upper.contains("REAL") || upper.contains("FLOA") || upper.contains("DOUB") {
177        (BaseDomain::F64, sim_relation_core::StorageRepr::F64)
178    } else {
179        (BaseDomain::Bytes, sim_relation_core::StorageRepr::Bytes)
180    }
181}
sim_platform_sqlite/driver/introspection.rs

sim_platform_sqlite/driver/
introspection.rs
