coven_core/sync/
session.rs

1//! Synced-table declarations and the shared identifier-quoting helper.
2//!
3//! [`SyncedTable`] is how a host declares which tables participate in changeset
4//! sync and what `(table, id)` means for each one. The set is no longer a
5//! process-global: the host passes it to
6//! [`crate::CovenBuilder::synced_tables`], and coven owns it for the lifetime of
7//! the connection and hands it to each journaled write's capture session, the
8//! gate, and apply.
9
10use std::collections::BTreeMap;
11
12use fallible_streaming_iterator::FallibleStreamingIterator;
13use rusqlite::hooks::Action;
14use rusqlite::session::{ChangesetItem, ChangesetIter};
15use rusqlite::types::ValueRef;
16use rusqlite::Connection;
17
18/// How `(table, id)` names one logical row across every device.
19///
20/// Equality always means one row, including equality between two valid UUIDs.
21/// The mode controls which ids may be introduced; it does not change merge
22/// equality. Changing a primary key removes the old identity and introduces the
23/// new one.
24#[derive(Debug, Clone, Copy, PartialEq, Eq)]
25pub enum RowIdentity {
26    /// Rows created independently use canonical lowercase hyphenated RFC UUID
27    /// version 4 or 7 ids.
28    IndependentUuid,
29    /// Application-assigned keys intentionally name the same logical row on
30    /// every device.
31    SharedKey,
32}
33
34#[derive(Debug, Clone, PartialEq, Eq, thiserror::Error)]
35pub(crate) enum RowIdentityError {
36    #[error(
37        "synced table {table:?} id {value:?} is invalid for IndependentUuid; expected a canonical lowercase hyphenated RFC UUID version 4 or 7"
38    )]
39    InvalidIndependentUuid { table: String, value: String },
40    #[error("synced table {table:?} changeset has no {side} primary-key value")]
41    MissingPrimaryKey { table: String, side: &'static str },
42    #[error("synced table {table:?} changeset has a non-TEXT {side} primary-key value")]
43    NonTextPrimaryKey { table: String, side: &'static str },
44    #[error("synced table {table:?} changeset primary key is not UTF-8: {reason}")]
45    NonUtf8PrimaryKey { table: String, reason: String },
46}
47
48impl RowIdentityError {
49    pub(crate) fn table(&self) -> &str {
50        match self {
51            Self::InvalidIndependentUuid { table, .. }
52            | Self::MissingPrimaryKey { table, .. }
53            | Self::NonTextPrimaryKey { table, .. }
54            | Self::NonUtf8PrimaryKey { table, .. } => table,
55        }
56    }
57}
58
59#[derive(Debug, thiserror::Error)]
60pub(crate) enum ChangesetIdentityError {
61    #[error("changeset row identity validation failed: {0}")]
62    Parse(String),
63    #[error("changeset contains undeclared table {0:?}")]
64    UndeclaredTable(String),
65    #[error(transparent)]
66    Row(#[from] RowIdentityError),
67}
68
69pub(crate) fn validate_row_identity(
70    table: &str,
71    identity: RowIdentity,
72    value: &str,
73) -> Result<(), RowIdentityError> {
74    if identity == RowIdentity::SharedKey {
75        return Ok(());
76    }
77
78    let valid = uuid::Uuid::parse_str(value).is_ok_and(|parsed| {
79        parsed.get_variant() == uuid::Variant::RFC4122
80            && matches!(
81                parsed.get_version(),
82                Some(uuid::Version::Random | uuid::Version::SortRand)
83            )
84            && parsed.hyphenated().to_string() == value
85    });
86    if valid {
87        Ok(())
88    } else {
89        Err(RowIdentityError::InvalidIndependentUuid {
90            table: table.to_string(),
91            value: value.to_string(),
92        })
93    }
94}
95
96pub(crate) fn validate_changeset_row_identities(
97    bytes: &[u8],
98    tables: &[SyncedTable],
99) -> Result<(), ChangesetIdentityError> {
100    if bytes.is_empty() {
101        return Ok(());
102    }
103
104    let input: &mut dyn std::io::Read = &mut &bytes[..];
105    let mut iter = ChangesetIter::start_strm(&input)
106        .map_err(|error| ChangesetIdentityError::Parse(error.to_string()))?;
107    while let Some(item) = iter
108        .next()
109        .map_err(|error| ChangesetIdentityError::Parse(error.to_string()))?
110    {
111        let op = item
112            .op()
113            .map_err(|error| ChangesetIdentityError::Parse(error.to_string()))?;
114        let table_name = op.table_name();
115        let table = tables
116            .iter()
117            .find(|table| table.name() == table_name)
118            .ok_or_else(|| ChangesetIdentityError::UndeclaredTable(table_name.to_string()))?;
119        match op.code() {
120            Action::SQLITE_INSERT => {
121                let id = required_changeset_id(item, table_name, "new", ChangesetSide::New)?;
122                validate_row_identity(table_name, table.row_identity(), &id)?;
123            }
124            Action::SQLITE_DELETE => {
125                let id = required_changeset_id(item, table_name, "old", ChangesetSide::Old)?;
126                validate_row_identity(table_name, table.row_identity(), &id)?;
127            }
128            Action::SQLITE_UPDATE => {
129                let old = required_changeset_id(item, table_name, "old", ChangesetSide::Old)?;
130                let id = optional_changeset_id(item, table_name, "new", ChangesetSide::New)?
131                    .unwrap_or(old);
132                validate_row_identity(table_name, table.row_identity(), &id)?;
133            }
134            _ => {}
135        }
136    }
137    Ok(())
138}
139
140#[derive(Clone, Copy)]
141enum ChangesetSide {
142    Old,
143    New,
144}
145
146fn required_changeset_id(
147    item: &ChangesetItem,
148    table: &str,
149    side_name: &'static str,
150    side: ChangesetSide,
151) -> Result<String, RowIdentityError> {
152    optional_changeset_id(item, table, side_name, side)?.ok_or_else(|| {
153        RowIdentityError::MissingPrimaryKey {
154            table: table.to_string(),
155            side: side_name,
156        }
157    })
158}
159
160fn optional_changeset_id(
161    item: &ChangesetItem,
162    table: &str,
163    side_name: &'static str,
164    side: ChangesetSide,
165) -> Result<Option<String>, RowIdentityError> {
166    let value = match side {
167        ChangesetSide::Old => item.old_value(0),
168        ChangesetSide::New => item.new_value(0),
169    };
170    let value = match value {
171        Ok(value) => value,
172        Err(rusqlite::Error::InvalidColumnIndex(_)) => return Ok(None),
173        Err(error) => {
174            return Err(RowIdentityError::NonUtf8PrimaryKey {
175                table: table.to_string(),
176                reason: error.to_string(),
177            })
178        }
179    };
180    let ValueRef::Text(bytes) = value else {
181        return Err(RowIdentityError::NonTextPrimaryKey {
182            table: table.to_string(),
183            side: side_name,
184        });
185    };
186    std::str::from_utf8(bytes)
187        .map(str::to_owned)
188        .map(Some)
189        .map_err(|error| RowIdentityError::NonUtf8PrimaryKey {
190            table: table.to_string(),
191            reason: error.to_string(),
192        })
193}
194
195/// A table that participates in changeset sync, declared at startup by the host
196/// and passed to [`crate::CovenBuilder::synced_tables`].
197///
198/// A plain [`SyncedTable::new`] table syncs unconditionally — every row goes to
199/// peers. [`SyncedTable::remote_root`] keeps that whole-table row sync and also
200/// makes the row a blob-locality root whose blobs are always Remote.
201/// [`SyncedTable::gated_by`] makes it a *gated root*: a boolean column whose
202/// truth decides, per row, whether that row (and its declared FK-descendants) is
203/// shared. A gated-false root and its subtree stay local; flipping the gate true
204/// re-emits the whole now-visible subtree to peers, and flipping it false again
205/// retracts that subtree from peers (emitting deletes for the rows leaving the
206/// shared set) while the rows stay local.
207///
208/// [`SyncedTable::gated_by_descendants`] is the upward complement: an
209/// always-shared *ancestor* that should sync only while at least one gated
210/// descendant survives. Without it, an album whose only releases are gated out
211/// would still sync its own row and land on peers as an orphan with zero
212/// children. A gated-by-descendants ancestor is cut exactly when its gated
213/// subtree is empty, and the keep composes recursively up the foreign-key chain
214/// (an artist syncs iff a surviving album references it, which syncs iff a
215/// surviving release does). The keep-children are *inferred* from the
216/// foreign-key graph, not declared — listing them by hand would restate the
217/// schema and drift the moment a new foreign key is added.
218///
219/// A table is *either* a remote root, a gated root, a gated-by-descendants
220/// ancestor, or plain — never two of these. See [`super::gate`] for the gating
221/// mechanics.
222/// Orthogonally, any table may *carry a blob* ([`SyncedTable::carries_blob`]):
223/// blob-bearing-ness is a property of the row's columns, not of its gate role.
224/// A table may also be marked an *asset* ([`SyncedTable::asset`]): a decoration
225/// (a cover, an artist image) that rides its foreign-key subject's gate but never
226/// keeps that subject alive. Asset-ness is likewise independent of the gate role.
227///
228/// Each table must have an `id` text primary key at column 0 and an
229/// `_updated_at TEXT NOT NULL` column (the HLC/LWW timestamp). Tables not in the
230/// set the host declares on the builder are local-only and never synced — that is
231/// also the mechanism for keeping device-local state (per-device pin/cache
232/// columns, local paths) out of sync: put it in a table you don't declare. An
233/// empty set is rejected by [`super::cycle::init_sync`].
234///
235/// The required [`RowIdentity`] defines which ids may name rows. Use
236/// [`RowIdentity::IndependentUuid`] for independently created rows and
237/// [`RowIdentity::SharedKey`] only when equal application keys intentionally
238/// name and merge as the same row.
239#[derive(Debug, Clone, PartialEq, Eq)]
240pub struct SyncedTable {
241    name: String,
242    row_identity: RowIdentity,
243    role: GateRole,
244    blob: Option<BlobDecl>,
245    /// Whether this table is an asset of its FK subject: it rides the subject's
246    /// gate as an inherited child but is excluded from the subject's
247    /// `gated_by_descendants` keep computation, so an asset row never keeps an
248    /// otherwise-empty ancestor alive. Orthogonal to [`GateRole`] and the blob.
249    asset: bool,
250}
251
252/// How a synced table relates to the gate. Orthogonal to whether it carries a
253/// blob.
254#[derive(Debug, Clone, PartialEq, Eq)]
255pub enum GateRole {
256    /// Every row syncs unconditionally.
257    Plain,
258    /// Every row syncs unconditionally, and blobs on the row or its descendants
259    /// are Remote by construction.
260    RemoteRoot,
261    /// A gated root: a row syncs iff its boolean `gate_column` is true, and the
262    /// gate flows down declared foreign keys to descendant rows.
263    GatedRoot { gate_column: String },
264    /// An audience root: NULL is Store, `local` is device-local, and every
265    /// other value is a canonical committed circle id.
266    ScopedRoot { audience_column: String },
267    /// An always-shared ancestor kept alive by its gated subtree: a row syncs
268    /// iff at least one foreign-key descendant table holds a surviving (kept)
269    /// row referencing it. A *marker* only; the keep-children are inferred from
270    /// the live foreign-key graph at gate-build time, never listed here.
271    GatedByDescendants,
272}
273
274impl SyncedTable {
275    /// An ungated synced table: every row syncs under the required identity
276    /// mode.
277    pub fn new(name: impl Into<String>, row_identity: RowIdentity) -> Self {
278        SyncedTable {
279            name: name.into(),
280            row_identity,
281            role: GateRole::Plain,
282            blob: None,
283            asset: false,
284        }
285    }
286
287    /// Make this a gated root: rows sync iff the boolean `column` is true.
288    pub fn gated_by(mut self, column: impl Into<String>) -> Self {
289        self.role = GateRole::GatedRoot {
290            gate_column: column.into(),
291        };
292        self
293    }
294
295    /// Make this an audience root whose TEXT column selects Store, Local, or
296    /// one committed circle for the row and its foreign-key descendants. A
297    /// store with an audience root requires [`crate::HomeStorage::Opaque`].
298    pub fn scoped_by(mut self, column: impl Into<String>) -> Self {
299        self.role = GateRole::ScopedRoot {
300            audience_column: column.into(),
301        };
302        self
303    }
304
305    /// Make this a remote root: every row syncs, and blobs on the row or its
306    /// foreign-key descendants are always Remote. There is no Local state for
307    /// [`crate::blob::transition::make_remote`] or
308    /// [`crate::blob::transition::make_local`] to transition.
309    pub fn remote_root(mut self) -> Self {
310        self.role = GateRole::RemoteRoot;
311        self
312    }
313
314    /// Make this an always-shared ancestor kept alive by its gated subtree: a
315    /// row syncs iff a surviving (kept) descendant row references it. The
316    /// keep-children are inferred from the foreign-key graph at gate-build time,
317    /// so there is nothing to pass here.
318    pub fn gated_by_descendants(mut self) -> Self {
319        self.role = GateRole::GatedByDescendants;
320        self
321    }
322
323    /// Declare that rows of this table carry a blob, located by the columns in
324    /// `decl`. coven derives the blob set itself from these columns + the live
325    /// schema (see [`crate::blob::decl::BlobDecls`]); it never calls back to the
326    /// host to discover blobs. Independent of the gate role.
327    pub fn carries_blob(mut self, decl: BlobDecl) -> Self {
328        self.blob = Some(decl);
329        self
330    }
331
332    /// Mark this table an *asset* of its FK subject: a host-provided decoration
333    /// (a cover, an artist image) that rides its subject's gate but never grants
334    /// keep. The asset still inherits the gate as a child of its subject — it
335    /// syncs exactly when the subject is kept — but the gate excludes it from the
336    /// subject's `gated_by_descendants` keep computation, so an asset row alone
337    /// never keeps an otherwise-empty ancestor alive (and the asset-rides-subject
338    /// vs. subject-kept-by-children relation can never form a cycle). Independent
339    /// of the gate role; declare it on an FK child of the subject.
340    pub fn asset(mut self) -> Self {
341        self.asset = true;
342        self
343    }
344
345    /// The table name.
346    pub fn name(&self) -> &str {
347        &self.name
348    }
349
350    /// How this table's ids name logical rows across devices.
351    pub fn row_identity(&self) -> RowIdentity {
352        self.row_identity
353    }
354
355    /// The gate column name, if this table is a gated root.
356    pub fn gate_column(&self) -> Option<&str> {
357        match &self.role {
358            GateRole::GatedRoot { gate_column } => Some(gate_column),
359            GateRole::Plain
360            | GateRole::RemoteRoot
361            | GateRole::ScopedRoot { .. }
362            | GateRole::GatedByDescendants => None,
363        }
364    }
365
366    /// The audience column name, if this table is a scoped root.
367    pub fn audience_column(&self) -> Option<&str> {
368        match &self.role {
369            GateRole::ScopedRoot { audience_column } => Some(audience_column),
370            GateRole::Plain
371            | GateRole::RemoteRoot
372            | GateRole::GatedRoot { .. }
373            | GateRole::GatedByDescendants => None,
374        }
375    }
376
377    /// The complete sync role included in the signed routing contract.
378    pub fn gate_role(&self) -> &GateRole {
379        &self.role
380    }
381
382    /// Whether this is a remote root: rows sync unconditionally, and blob
383    /// locality for the row and descendants is always Remote.
384    pub fn is_remote_root(&self) -> bool {
385        matches!(self.role, GateRole::RemoteRoot)
386    }
387
388    /// Whether this is a gated-by-descendants ancestor (kept alive by its gated
389    /// subtree rather than by a column of its own).
390    pub fn is_gated_by_descendants(&self) -> bool {
391        matches!(self.role, GateRole::GatedByDescendants)
392    }
393
394    /// This table's blob declaration, if it carries one.
395    pub fn blob(&self) -> Option<&BlobDecl> {
396        self.blob.as_ref()
397    }
398
399    /// Whether this table is an asset of its FK subject (rides the subject's gate
400    /// but never grants keep). See [`SyncedTable::asset`].
401    pub fn is_asset(&self) -> bool {
402        self.asset
403    }
404}
405
406/// Where a blob-bearing table's blob columns live, declared by the host so coven
407/// can derive every blob a row references without a runtime callback. Resolved
408/// against the live schema into a [`crate::blob::decl::BlobDecls`] each cycle.
409///
410/// A blob declares two orthogonal properties: [`provenance`](BlobDecl::provenance)
411/// (its Local story) and [`fill`](BlobDecl::fill) (its Remote story).
412#[derive(Debug, Clone, PartialEq, Eq)]
413pub struct BlobDecl {
414    /// The column holding the blob id. Defaults to the primary key (`id`, column
415    /// 0), which is the blob id for most tables.
416    pub id_column: String,
417    /// The column holding the blob's plaintext length in bytes.
418    pub size_column: String,
419    /// The column holding the blob's content hash — the lowercase-hex SHA-256 of
420    /// its plaintext, computed at import (see [`crate::blob::content_hash`]). The
421    /// row carries it in a signed changeset, so it is signed by the row's author;
422    /// on download coven hashes the decrypted plaintext and requires equality with
423    /// this value, so the bytes are pinned by the author, not by where they were
424    /// found. Defaults to `hash`.
425    pub hash_column: String,
426    /// Cloud namespace for the blob, e.g. `"images"` or `"audio"`.
427    pub namespace: String,
428    /// The column holding the consumer's readable cloud-relative path, used as the
429    /// object key under the plain (browsable) blob-path scheme. `None` means the
430    /// blob is keyed only by its hashed id (the default obfuscated scheme).
431    pub cloud_path_column: Option<String>,
432    /// How the blob is scoped for encryption (see [`crate::blob::BlobScope`]).
433    pub scope: crate::blob::BlobScope,
434    /// The blob's **Local story**: [`crate::blob::Provenance::UserProvided`] (the
435    /// user's file at a path) or [`crate::blob::Provenance::HostProvided`] (coven's
436    /// own copy in the local store).
437    pub provenance: crate::blob::Provenance,
438    /// The blob's **Remote story**: [`crate::blob::CacheFill::CacheEager`] (fetched
439    /// into the cache on every pull) or [`crate::blob::CacheFill::CacheLazy`]
440    /// (fetched into the cache on first read).
441    pub fill: crate::blob::CacheFill,
442    /// The blob's **replacement story**: whether this row may be repointed at a
443    /// different blob ([`crate::blob::BlobReplacement`]). Decides what coven requires of
444    /// the blob's cloud key so that a cloud object is never rewritten with different
445    /// bytes. Defaults to [`crate::blob::BlobReplacement::Replaceable`].
446    pub replacement: crate::blob::BlobReplacement,
447}
448
449impl BlobDecl {
450    /// A blob declaration in `namespace` with the given `provenance` (its Local
451    /// story) and cache `fill` (its Remote story), the blob id taken from the
452    /// primary key (`id`), no readable cloud path, master-scoped, and
453    /// [`Replaceable`](crate::blob::BlobReplacement::Replaceable). Refine with the
454    /// `with_*` builders.
455    pub fn new(
456        namespace: impl Into<String>,
457        provenance: crate::blob::Provenance,
458        fill: crate::blob::CacheFill,
459    ) -> Self {
460        BlobDecl {
461            id_column: "id".to_string(),
462            size_column: "size".to_string(),
463            hash_column: "hash".to_string(),
464            namespace: namespace.into(),
465            cloud_path_column: None,
466            scope: crate::blob::BlobScope::Master,
467            provenance,
468            fill,
469            replacement: crate::blob::BlobReplacement::Replaceable,
470        }
471    }
472
473    /// Declare that this table's row is never repointed at a different blob
474    /// ([`crate::blob::BlobReplacement::WriteOnce`]), which frees its readable
475    /// `cloud_path` to be a stable, fully human-readable name. coven refuses a
476    /// repointing. Read that variant's docs before reaching for this: it is a weaker
477    /// contract than the default, and it asks the consumer to guarantee the part coven
478    /// cannot see — that a path is never reused by a different blob.
479    pub fn write_once(mut self) -> Self {
480        self.replacement = crate::blob::BlobReplacement::WriteOnce;
481        self
482    }
483
484    /// Take the blob id from `column` instead of the primary key.
485    pub fn with_id_column(mut self, column: impl Into<String>) -> Self {
486        self.id_column = column.into();
487        self
488    }
489
490    /// Take the plaintext byte length from `column` instead of `size`.
491    pub fn with_size_column(mut self, column: impl Into<String>) -> Self {
492        self.size_column = column.into();
493        self
494    }
495
496    /// Take the content hash from `column` instead of `hash`.
497    pub fn with_hash_column(mut self, column: impl Into<String>) -> Self {
498        self.hash_column = column.into();
499        self
500    }
501
502    /// Key the blob at the readable cloud path in `column` (the plain scheme).
503    pub fn with_cloud_path_column(mut self, column: impl Into<String>) -> Self {
504        self.cloud_path_column = Some(column.into());
505        self
506    }
507
508    /// Scope the blob's encryption (defaults to [`crate::blob::BlobScope::Master`]).
509    pub fn with_scope(mut self, scope: crate::blob::BlobScope) -> Self {
510        self.scope = scope;
511        self
512    }
513}
514
515/// Column names of `table`, in declared order, via `PRAGMA table_info`. The
516/// index of a name here is the index SQLite session changesets report for that
517/// column.
518pub(crate) fn table_columns(conn: &Connection, table: &str) -> rusqlite::Result<Vec<String>> {
519    let sql = format!("PRAGMA table_info({})", quote_ident(table));
520    let mut stmt = conn.prepare(&sql)?;
521    let columns = stmt
522        .query_map([], |row| row.get::<_, String>(1))?
523        .collect::<Result<Vec<_>, _>>()?;
524    Ok(columns)
525}
526
527/// Quote an SQL identifier (table/column name), doubling any embedded quote, so
528/// a trusted-but-unbindable name interpolates safely. Identifiers cannot be
529/// passed as bound parameters; this is the safe interpolation path for them.
530pub(crate) fn quote_ident(ident: &str) -> String {
531    format!("\"{}\"", ident.replace('"', "\"\""))
532}
533
534#[derive(Debug, thiserror::Error)]
535pub(crate) enum ForeignKeySchemaError {
536    #[error(transparent)]
537    Sqlite(#[from] rusqlite::Error),
538    #[error("foreign key on {child_table:?} is malformed: {reason}")]
539    Malformed { child_table: String, reason: String },
540}
541
542#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord)]
543pub(crate) struct ForeignKeyColumn {
544    pub(crate) child: String,
545    pub(crate) parent: String,
546}
547
548#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord)]
549pub(crate) struct ForeignKeyEdge {
550    pub(crate) parent_table: String,
551    pub(crate) columns: Vec<ForeignKeyColumn>,
552    pub(crate) on_update: String,
553    pub(crate) on_delete: String,
554    pub(crate) match_clause: String,
555}
556
557struct ForeignKeyRow {
558    sequence: i64,
559    parent_table: String,
560    child_column: String,
561    parent_column: Option<String>,
562    on_update: String,
563    on_delete: String,
564    match_clause: String,
565}
566
567/// Every outgoing foreign key on `child_table`, grouped by constraint and
568/// sorted by its complete parent/column/action shape. SQLite's constraint ids
569/// and PRAGMA listing order are deliberately discarded.
570pub(crate) fn foreign_key_edges(
571    conn: &Connection,
572    child_table: &str,
573) -> Result<Vec<ForeignKeyEdge>, ForeignKeySchemaError> {
574    let sql = format!("PRAGMA foreign_key_list({})", quote_ident(child_table));
575    let mut statement = conn.prepare(&sql)?;
576    let rows = statement.query_map([], |row| {
577        Ok((
578            row.get::<_, i64>(0)?,
579            ForeignKeyRow {
580                sequence: row.get(1)?,
581                parent_table: row.get(2)?,
582                child_column: row.get(3)?,
583                parent_column: row.get(4)?,
584                on_update: row.get::<_, String>(5)?.to_ascii_uppercase(),
585                on_delete: row.get::<_, String>(6)?.to_ascii_uppercase(),
586                match_clause: row.get::<_, String>(7)?.to_ascii_uppercase(),
587            },
588        ))
589    })?;
590    let mut grouped: BTreeMap<i64, Vec<ForeignKeyRow>> = BTreeMap::new();
591    for row in rows {
592        let (id, row) = row?;
593        grouped.entry(id).or_default().push(row);
594    }
595
596    let mut edges = Vec::with_capacity(grouped.len());
597    for mut rows in grouped.into_values() {
598        rows.sort_by_key(|row| row.sequence);
599        let first = rows
600            .first()
601            .ok_or_else(|| ForeignKeySchemaError::Malformed {
602                child_table: child_table.to_string(),
603                reason: "constraint has no columns".to_string(),
604            })?;
605        if rows.iter().any(|row| {
606            row.parent_table != first.parent_table
607                || row.on_update != first.on_update
608                || row.on_delete != first.on_delete
609                || row.match_clause != first.match_clause
610        }) {
611            return Err(ForeignKeySchemaError::Malformed {
612                child_table: child_table.to_string(),
613                reason: "one constraint reports inconsistent parent or actions".to_string(),
614            });
615        }
616        let omitted_parent_columns = rows.iter().all(|row| row.parent_column.is_none());
617        if !omitted_parent_columns && rows.iter().any(|row| row.parent_column.is_none()) {
618            return Err(ForeignKeySchemaError::Malformed {
619                child_table: child_table.to_string(),
620                reason: "one constraint mixes named and omitted parent columns".to_string(),
621            });
622        }
623        let inferred_parent_columns = if omitted_parent_columns {
624            primary_key_columns(conn, &first.parent_table)?
625        } else {
626            Vec::new()
627        };
628        if omitted_parent_columns && inferred_parent_columns.len() != rows.len() {
629            return Err(ForeignKeySchemaError::Malformed {
630                child_table: child_table.to_string(),
631                reason: format!(
632                    "{} child columns reference {} primary-key columns",
633                    rows.len(),
634                    inferred_parent_columns.len(),
635                ),
636            });
637        }
638        let columns = rows
639            .iter()
640            .enumerate()
641            .map(|(position, row)| ForeignKeyColumn {
642                child: row.child_column.clone(),
643                parent: row
644                    .parent_column
645                    .clone()
646                    .unwrap_or_else(|| inferred_parent_columns[position].clone()),
647            })
648            .collect();
649        edges.push(ForeignKeyEdge {
650            parent_table: first.parent_table.clone(),
651            columns,
652            on_update: first.on_update.clone(),
653            on_delete: first.on_delete.clone(),
654            match_clause: first.match_clause.clone(),
655        });
656    }
657    edges.sort();
658    Ok(edges)
659}
660
661fn primary_key_columns(
662    conn: &Connection,
663    table: &str,
664) -> Result<Vec<String>, ForeignKeySchemaError> {
665    let sql = format!("PRAGMA table_info({})", quote_ident(table));
666    let mut statement = conn.prepare(&sql)?;
667    let rows = statement.query_map([], |row| {
668        Ok((row.get::<_, i64>(5)?, row.get::<_, String>(1)?))
669    })?;
670    let mut columns = rows
671        .collect::<rusqlite::Result<Vec<_>>>()?
672        .into_iter()
673        .filter(|(rank, _)| *rank > 0)
674        .collect::<Vec<_>>();
675    columns.sort_by_key(|(rank, _)| *rank);
676    Ok(columns.into_iter().map(|(_, name)| name).collect())
677}
678
679#[cfg(test)]
680mod row_identity_tests {
681    use super::*;
682
683    #[test]
684    fn independent_uuid_accepts_only_canonical_rfc_uuid_v4_or_v7() {
685        for valid in [
686            "f47ac10b-58cc-4372-a567-0e02b2c3d479",
687            "01890a5d-ac96-774b-bcce-b302099c3f74",
688        ] {
689            validate_row_identity("things", RowIdentity::IndependentUuid, valid)
690                .unwrap_or_else(|error| panic!("{valid} must be accepted: {error}"));
691        }
692
693        for invalid in [
694            "F47AC10B-58CC-4372-A567-0E02B2C3D479",
695            "f47ac10b58cc4372a5670e02b2c3d479",
696            "{f47ac10b-58cc-4372-a567-0e02b2c3d479}",
697            "urn:uuid:f47ac10b-58cc-4372-a567-0e02b2c3d479",
698            "00000000-0000-0000-0000-000000000000",
699            "f47ac10b-58cc-1372-a567-0e02b2c3d479",
700            "f47ac10b-58cc-2372-a567-0e02b2c3d479",
701            "f47ac10b-58cc-3372-a567-0e02b2c3d479",
702            "f47ac10b-58cc-5372-a567-0e02b2c3d479",
703            "f47ac10b-58cc-6372-a567-0e02b2c3d479",
704            "f47ac10b-58cc-8372-a567-0e02b2c3d479",
705            "f47ac10b-58cc-4372-0567-0e02b2c3d479",
706            "not-a-uuid",
707        ] {
708            assert!(
709                validate_row_identity("things", RowIdentity::IndependentUuid, invalid).is_err(),
710                "{invalid} must be rejected",
711            );
712        }
713
714        validate_row_identity("settings", RowIdentity::SharedKey, "preferences")
715            .expect("shared keys accept application-assigned ids");
716    }
717}
coven_core/sync/session.rs

coven_core/sync/
session.rs
