coven_database/
schema_contract.rs

1use super::*;
2use crate::query_mapped_rows;
3use std::collections::HashMap;
4
5#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
6#[serde(deny_unknown_fields)]
7pub struct DurablePreparedProtocolObject {
8    pub semantic_bytes: Vec<u8>,
9    pub prepared: PreparedExactObject,
10}
11
12impl DurablePreparedProtocolObject {
13    pub fn new(semantic_bytes: Vec<u8>, prepared: PreparedExactObject) -> Self {
14        Self {
15            semantic_bytes,
16            prepared,
17        }
18    }
19
20    pub fn semantic_bytes(&self) -> &[u8] {
21        &self.semantic_bytes
22    }
23
24    pub fn prepared(&self) -> &PreparedExactObject {
25        &self.prepared
26    }
27}
28
29#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
30#[serde(deny_unknown_fields)]
31pub struct StoreBatchLocalCleanup {
32    pub drops: Vec<coven_protocol::blob::DeferredLocalBlobDrop>,
33}
34
35#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
36#[serde(deny_unknown_fields)]
37pub struct StoreBatchCompletion {}
38
39pub(crate) fn validate_host_synced_tables(
40    conn: &Connection,
41    synced_tables: &[SyncedTable],
42) -> Result<(), DbError> {
43    validate_host_trigger_names(conn)?;
44
45    // Which table owns each declared blob namespace, so a second table claiming an
46    // already-owned namespace is caught here. A blob's namespace is part of its
47    // address; two tables sharing one makes `row_for_blob_in_namespace` resolve to
48    // whichever the hash map iterates first.
49    let mut namespace_owner: HashMap<&str, &str> = HashMap::new();
50    let mut table_by_sqlite_name: HashMap<String, &str> = HashMap::new();
51    for table in synced_tables {
52        let name = table.name();
53        if name.is_empty() {
54            return Err(DbError::Message(
55                "synced table name must not be empty".to_string(),
56            ));
57        }
58        if is_reserved_table_name(name) {
59            return Err(DbError::Message(format!(
60                "synced table {name:?} is reserved by coven"
61            )));
62        }
63        let sqlite_name = name.to_ascii_lowercase();
64        if let Some(prior) = table_by_sqlite_name.insert(sqlite_name, name) {
65            return Err(DbError::Message(format!(
66                "synced tables {prior:?} and {name:?} are declared as the same SQLite table more than once"
67            )));
68        }
69        if let Some(live_name) = canonical_table_name(conn, name)? {
70            if live_name != name {
71                return Err(DbError::Message(format!(
72                    "synced table {name:?} does not use the live schema's exact spelling {live_name:?}"
73                )));
74            }
75        }
76        validate_synced_table_contract(conn, name)?;
77        validate_existing_row_identities(conn, table)?;
78        if let Some(decl) = table.blob() {
79            let namespace = decl.namespace.as_str();
80            if let Some(prior) = namespace_owner.insert(namespace, name) {
81                return Err(DbError::Message(format!(
82                    "synced tables {prior:?} and {name:?} both declare blob namespace \
83                     {namespace:?}; a namespace must be owned by exactly one table"
84                )));
85            }
86        }
87    }
88    Ok(())
89}
90
91fn validate_host_trigger_names(conn: &Connection) -> Result<(), DbError> {
92    let names = query_mapped_rows(
93        conn,
94        "SELECT name FROM main.sqlite_schema WHERE type = 'trigger'
95             UNION ALL
96             SELECT name FROM temp.sqlite_schema WHERE type = 'trigger'",
97        [],
98        |row| row.get::<_, String>(0),
99    )?;
100    for name in names {
101        if is_coven_cleanup_guard_name(&name) {
102            return Err(DbError::Message(format!(
103                "host trigger {name:?} uses a name reserved for Coven blob cleanup guards"
104            )));
105        }
106    }
107    Ok(())
108}
109
110/// Return the live `main`-schema spelling SQLite resolves for `table`.
111/// SQLite table identifiers compare case-insensitively, while coven dispatches
112/// changesets by their exact table name, so open requires declarations to use
113/// this canonical spelling.
114pub(crate) fn canonical_table_name(
115    conn: &Connection,
116    table: &str,
117) -> Result<Option<String>, DbError> {
118    conn.query_row(
119        "SELECT name FROM main.sqlite_schema \
120         WHERE type = 'table' AND name = ?1 COLLATE NOCASE",
121        [table],
122        |row| row.get(0),
123    )
124    .optional()
125    .map_err(DbError::from)
126}
127
128pub(crate) fn validate_existing_row_identities(
129    conn: &Connection,
130    table: &SyncedTable,
131) -> Result<(), DbError> {
132    if table.row_identity() == coven_protocol::synced_schema::RowIdentity::SharedKey {
133        return Ok(());
134    }
135    let sql = format!("SELECT id FROM {}", crate::quote_ident(table.name()));
136    let ids = query_mapped_rows(conn, &sql, [], |row| row.get::<_, String>(0))?;
137    for id in ids {
138        table
139            .row_identity()
140            .validate(table.name(), &id)
141            .map_err(DbError::from)?;
142    }
143    Ok(())
144}
145
146/// One column of a table's `PRAGMA table_info`. `position` is the column ordinal
147/// — the index a session changeset reports for that column, so the pk's position
148/// is what the by-position apply path reads. `pk` is 0 for a non-key column or its
149/// 1-based rank within the primary key.
150pub(crate) struct ColumnInfo {
151    position: i64,
152    name: String,
153    declared_type: String,
154    not_null: bool,
155    pk: i64,
156}
157
158/// Enforce the synced-table contract ([`coven_protocol::synced_schema::SyncedTable`]) on
159/// `table`'s live schema: the table declared STRICT; a single primary key
160/// column, named `id`, declared TEXT, at column 0; and an `_updated_at` column
161/// declared TEXT NOT NULL. A violation is an open error naming the table and the
162/// requirement it broke, so the integrator learns it on their own device instead
163/// of a peer's pull failing on the row.
164pub(crate) fn validate_synced_table_contract(
165    conn: &Connection,
166    table: &str,
167) -> Result<(), DbError> {
168    match table_is_strict(conn, table)? {
169        None => {
170            return Err(DbError::Message(format!(
171                "synced table {table:?} is declared in `synced_tables` but no migration \
172                 creates it — add a `CREATE TABLE {table} (...) STRICT` to the schema \
173                 migrations, or remove the declaration"
174            )));
175        }
176        Some(false) => {
177            return Err(DbError::Message(format!(
178                "synced table {table:?} is not declared STRICT; the sync contract assumes typed \
179                 columns (apply preserves storage classes peer-to-peer, LWW arbitration renders \
180                 values to strings for comparison), which STRICT enforces at the insert — declare \
181                 it STRICT: `CREATE TABLE {table} (...) STRICT`"
182            )));
183        }
184        Some(true) => {}
185    }
186
187    let sql = format!("PRAGMA table_info({})", crate::quote_ident(table));
188    let mut stmt = conn.prepare(&sql).map_err(DbError::from)?;
189    let mut columns = Vec::new();
190    let rows = stmt
191        .query_map([], |row| {
192            Ok(ColumnInfo {
193                position: row.get::<_, i64>(0)?,
194                name: row.get::<_, String>(1)?,
195                declared_type: row.get::<_, String>(2)?,
196                not_null: row.get::<_, i64>(3)? != 0,
197                pk: row.get::<_, i64>(5)?,
198            })
199        })
200        .map_err(DbError::from)?;
201    for row in rows {
202        columns.push(row.map_err(DbError::from)?);
203    }
204
205    let pk_columns: Vec<&ColumnInfo> = columns.iter().filter(|c| c.pk > 0).collect();
206    let pk = match pk_columns.as_slice() {
207        [single] => *single,
208        [] => {
209            return Err(DbError::Message(format!(
210                "synced table {table:?} has no primary key; the contract requires a single \
211                 `id` TEXT primary key at column 0"
212            )))
213        }
214        _ => {
215            let names: Vec<&str> = pk_columns.iter().map(|c| c.name.as_str()).collect();
216            return Err(DbError::Message(format!(
217                "synced table {table:?} has a composite primary key {names:?}; the contract \
218                 requires a single `id` TEXT primary key at column 0"
219            )));
220        }
221    };
222    if pk.name != "id" {
223        return Err(DbError::Message(format!(
224            "synced table {table:?} primary key is {:?}, not `id`; the contract requires the \
225             primary key to be the `id` column",
226            pk.name
227        )));
228    }
229    if pk.position != 0 {
230        return Err(DbError::Message(format!(
231            "synced table {table:?} primary key `id` is at column {}, not column 0; the \
232             contract requires `id` to be the first column",
233            pk.position
234        )));
235    }
236    if !declared_as_text(&pk.declared_type) {
237        return Err(DbError::Message(format!(
238            "synced table {table:?} primary key `id` is declared {:?}, not TEXT; the contract \
239             requires an `id` TEXT primary key",
240            pk.declared_type
241        )));
242    }
243
244    let updated_at = columns
245        .iter()
246        .find(|c| c.name == "_updated_at")
247        .ok_or_else(|| {
248            DbError::Message(format!(
249                "synced table {table:?} has no `_updated_at` column; the contract requires \
250                 `_updated_at TEXT NOT NULL`"
251            ))
252        })?;
253    if !declared_as_text(&updated_at.declared_type) {
254        return Err(DbError::Message(format!(
255            "synced table {table:?} column `_updated_at` is declared {:?}, not TEXT; the \
256             contract requires `_updated_at TEXT NOT NULL`",
257            updated_at.declared_type
258        )));
259    }
260    if !updated_at.not_null {
261        return Err(DbError::Message(format!(
262            "synced table {table:?} column `_updated_at` is nullable; the contract requires \
263             `_updated_at TEXT NOT NULL`"
264        )));
265    }
266
267    Ok(())
268}
269
270/// Whether a `PRAGMA table_info` declared type is TEXT, case-insensitively. SQL
271/// keywords are case-insensitive, so `text` and `TEXT` both satisfy the contract;
272/// any other declared type (or none) does not.
273pub(crate) fn declared_as_text(declared_type: &str) -> bool {
274    declared_type.eq_ignore_ascii_case("TEXT")
275}
276
277/// Whether `table` (in the `main` schema) is declared STRICT, via `PRAGMA
278/// table_list`'s `strict` column (SQLite 3.37+) — the schema-level flag itself,
279/// not `sqlite_master.sql` text, which a hand-formatted `CREATE TABLE` could spell
280/// many ways. `None` means the table doesn't exist in `main` at all — a declared
281/// synced table no migration created — which the caller reports as its own
282/// contract error rather than folding into "not STRICT".
283pub(crate) fn table_is_strict(conn: &Connection, table: &str) -> Result<Option<bool>, DbError> {
284    let sql = format!("PRAGMA table_list({})", crate::quote_ident(table));
285    let rows = query_mapped_rows(conn, &sql, [], |row| {
286        Ok((row.get::<_, String>(0)?, row.get::<_, i64>(5)?))
287    })?;
288    for (schema, strict) in rows {
289        if schema == "main" {
290            return Ok(Some(strict != 0));
291        }
292    }
293    Ok(None)
294}
coven_database/schema_contract.rs

coven_database/
schema_contract.rs
