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    audience_parent_column: Option<String>,
245    blob: Option<BlobDecl>,
246    /// Whether this table is an asset of its FK subject: it rides the subject's
247    /// gate as an inherited child but is excluded from the subject's
248    /// `gated_by_descendants` keep computation, so an asset row never keeps an
249    /// otherwise-empty ancestor alive. Orthogonal to [`GateRole`] and the blob.
250    asset: bool,
251}
252
253/// How a synced table relates to the gate. Orthogonal to whether it carries a
254/// blob.
255#[derive(Debug, Clone, PartialEq, Eq)]
256pub enum GateRole {
257    /// Every row syncs unconditionally.
258    Plain,
259    /// Every row syncs unconditionally, and blobs on the row or its descendants
260    /// are Remote by construction.
261    RemoteRoot,
262    /// A gated root: a row syncs iff its boolean `gate_column` is true, and the
263    /// gate flows down declared foreign keys to descendant rows.
264    GatedRoot { gate_column: String },
265    /// An audience root: NULL is Store, `local` is device-local, and every
266    /// other value is a canonical committed circle id.
267    ScopedRoot { audience_column: String },
268    /// An always-shared ancestor kept alive by its gated subtree: a row syncs
269    /// iff at least one foreign-key descendant table holds a surviving (kept)
270    /// row referencing it. A *marker* only; the keep-children are inferred from
271    /// the live foreign-key graph at gate-build time, never listed here.
272    GatedByDescendants,
273}
274
275impl SyncedTable {
276    /// An ungated synced table: every row syncs under the required identity
277    /// mode.
278    pub fn new(name: impl Into<String>, row_identity: RowIdentity) -> Self {
279        SyncedTable {
280            name: name.into(),
281            row_identity,
282            role: GateRole::Plain,
283            audience_parent_column: None,
284            blob: None,
285            asset: false,
286        }
287    }
288
289    /// Make this a gated root: rows sync iff the boolean `column` is true.
290    pub fn gated_by(mut self, column: impl Into<String>) -> Self {
291        self.role = GateRole::GatedRoot {
292            gate_column: column.into(),
293        };
294        self
295    }
296
297    /// Make this an audience root whose TEXT column selects Store, Local, or
298    /// one committed circle for the row and its foreign-key descendants. A
299    /// store with an audience root requires [`crate::HomeStorage::Opaque`].
300    pub fn scoped_by(mut self, column: impl Into<String>) -> Self {
301        self.role = GateRole::ScopedRoot {
302            audience_column: column.into(),
303        };
304        self
305    }
306
307    /// Make this plain table inherit its audience through the foreign key whose
308    /// child column is `column`. Every descendant of an audience root must
309    /// select this relationship explicitly; coven never guesses among the
310    /// table's foreign keys.
311    pub fn inherits_audience_through(mut self, column: impl Into<String>) -> Self {
312        self.audience_parent_column = Some(column.into());
313        self
314    }
315
316    /// Make this a remote root: every row syncs, and blobs on the row or its
317    /// foreign-key descendants are always Remote. There is no Local state for
318    /// [`crate::blob::transition::make_remote`] or
319    /// [`crate::blob::transition::make_local`] to transition.
320    pub fn remote_root(mut self) -> Self {
321        self.role = GateRole::RemoteRoot;
322        self
323    }
324
325    /// Make this an always-shared ancestor kept alive by its gated subtree: a
326    /// row syncs iff a surviving (kept) descendant row references it. The
327    /// keep-children are inferred from the foreign-key graph at gate-build time,
328    /// so there is nothing to pass here.
329    pub fn gated_by_descendants(mut self) -> Self {
330        self.role = GateRole::GatedByDescendants;
331        self
332    }
333
334    /// Declare that rows of this table carry a blob, located by the columns in
335    /// `decl`. coven derives the blob set itself from these columns + the live
336    /// schema (see [`crate::blob::decl::BlobDecls`]); it never calls back to the
337    /// host to discover blobs. Independent of the gate role.
338    pub fn carries_blob(mut self, decl: BlobDecl) -> Self {
339        self.blob = Some(decl);
340        self
341    }
342
343    /// Mark this table an *asset* of its FK subject: a host-provided decoration
344    /// (a cover, an artist image) that rides its subject's gate but never grants
345    /// keep. The asset still inherits the gate as a child of its subject — it
346    /// syncs exactly when the subject is kept — but the gate excludes it from the
347    /// subject's `gated_by_descendants` keep computation, so an asset row alone
348    /// never keeps an otherwise-empty ancestor alive (and the asset-rides-subject
349    /// vs. subject-kept-by-children relation can never form a cycle). Independent
350    /// of the gate role; declare it on an FK child of the subject.
351    pub fn asset(mut self) -> Self {
352        self.asset = true;
353        self
354    }
355
356    /// The table name.
357    pub fn name(&self) -> &str {
358        &self.name
359    }
360
361    /// How this table's ids name logical rows across devices.
362    pub fn row_identity(&self) -> RowIdentity {
363        self.row_identity
364    }
365
366    /// The gate column name, if this table is a gated root.
367    pub fn gate_column(&self) -> Option<&str> {
368        match &self.role {
369            GateRole::GatedRoot { gate_column } => Some(gate_column),
370            GateRole::Plain
371            | GateRole::RemoteRoot
372            | GateRole::ScopedRoot { .. }
373            | GateRole::GatedByDescendants => None,
374        }
375    }
376
377    /// The audience column name, if this table is a scoped root.
378    pub fn audience_column(&self) -> Option<&str> {
379        match &self.role {
380            GateRole::ScopedRoot { audience_column } => Some(audience_column),
381            GateRole::Plain
382            | GateRole::RemoteRoot
383            | GateRole::GatedRoot { .. }
384            | GateRole::GatedByDescendants => None,
385        }
386    }
387
388    /// The complete sync role included in the signed routing contract.
389    pub fn gate_role(&self) -> &GateRole {
390        &self.role
391    }
392
393    /// The child foreign-key column that selects this table's audience parent.
394    pub fn audience_parent_column(&self) -> Option<&str> {
395        self.audience_parent_column.as_deref()
396    }
397
398    /// Whether this is a remote root: rows sync unconditionally, and blob
399    /// locality for the row and descendants is always Remote.
400    pub fn is_remote_root(&self) -> bool {
401        matches!(self.role, GateRole::RemoteRoot)
402    }
403
404    /// Whether this is a gated-by-descendants ancestor (kept alive by its gated
405    /// subtree rather than by a column of its own).
406    pub fn is_gated_by_descendants(&self) -> bool {
407        matches!(self.role, GateRole::GatedByDescendants)
408    }
409
410    /// This table's blob declaration, if it carries one.
411    pub fn blob(&self) -> Option<&BlobDecl> {
412        self.blob.as_ref()
413    }
414
415    /// Whether this table is an asset of its FK subject (rides the subject's gate
416    /// but never grants keep). See [`SyncedTable::asset`].
417    pub fn is_asset(&self) -> bool {
418        self.asset
419    }
420}
421
422/// Where a blob-bearing table's blob columns live, declared by the host so coven
423/// can derive every blob a row references without a runtime callback. Resolved
424/// against the live schema into a [`crate::blob::decl::BlobDecls`] each cycle.
425///
426/// A blob declares two orthogonal properties: [`provenance`](BlobDecl::provenance)
427/// (its Local story) and [`fill`](BlobDecl::fill) (its Remote story).
428#[derive(Debug, Clone, PartialEq, Eq)]
429pub struct BlobDecl {
430    /// The column holding the blob id. Defaults to the primary key (`id`, column
431    /// 0), which is the blob id for most tables.
432    pub id_column: String,
433    /// The column holding the blob's plaintext length in bytes.
434    pub size_column: String,
435    /// The column holding the blob's content hash — the lowercase-hex SHA-256 of
436    /// its plaintext, computed at import (see [`crate::blob::content_hash`]). The
437    /// row carries it in a signed changeset, so it is signed by the row's author;
438    /// on download coven hashes the decrypted plaintext and requires equality with
439    /// this value, so the bytes are pinned by the author, not by where they were
440    /// found. Defaults to `hash`.
441    pub hash_column: String,
442    /// Cloud namespace for the blob, e.g. `"images"` or `"audio"`.
443    pub namespace: String,
444    /// The column holding the consumer's readable cloud-relative path, used as the
445    /// object key under the plain (browsable) blob-path scheme. `None` means the
446    /// blob is keyed only by its hashed id (the default obfuscated scheme).
447    pub cloud_path_column: Option<String>,
448    /// How the blob is scoped for encryption (see [`crate::blob::BlobScope`]).
449    pub scope: crate::blob::BlobScope,
450    /// The blob's **Local story**: [`crate::blob::Provenance::UserProvided`] (the
451    /// user's file at a path) or [`crate::blob::Provenance::HostProvided`] (coven's
452    /// own copy in the local store).
453    pub provenance: crate::blob::Provenance,
454    /// The blob's **Remote story**: [`crate::blob::CacheFill::CacheEager`] (fetched
455    /// into the cache on every pull) or [`crate::blob::CacheFill::CacheLazy`]
456    /// (fetched into the cache on first read).
457    pub fill: crate::blob::CacheFill,
458    /// The blob's **replacement story**: whether this row may be repointed at a
459    /// different blob ([`crate::blob::BlobReplacement`]). Decides what coven requires of
460    /// the blob's cloud key so that a cloud object is never rewritten with different
461    /// bytes. Defaults to [`crate::blob::BlobReplacement::Replaceable`].
462    pub replacement: crate::blob::BlobReplacement,
463}
464
465impl BlobDecl {
466    /// A blob declaration in `namespace` with the given `provenance` (its Local
467    /// story) and cache `fill` (its Remote story), the blob id taken from the
468    /// primary key (`id`), no readable cloud path, master-scoped, and
469    /// [`Replaceable`](crate::blob::BlobReplacement::Replaceable). Refine with the
470    /// `with_*` builders.
471    pub fn new(
472        namespace: impl Into<String>,
473        provenance: crate::blob::Provenance,
474        fill: crate::blob::CacheFill,
475    ) -> Self {
476        BlobDecl {
477            id_column: "id".to_string(),
478            size_column: "size".to_string(),
479            hash_column: "hash".to_string(),
480            namespace: namespace.into(),
481            cloud_path_column: None,
482            scope: crate::blob::BlobScope::Master,
483            provenance,
484            fill,
485            replacement: crate::blob::BlobReplacement::Replaceable,
486        }
487    }
488
489    /// Declare that this table's row is never repointed at a different blob
490    /// ([`crate::blob::BlobReplacement::WriteOnce`]), which frees its readable
491    /// `cloud_path` to be a stable, fully human-readable name. coven refuses a
492    /// repointing. Read that variant's docs before reaching for this: it is a weaker
493    /// contract than the default, and it asks the consumer to guarantee the part coven
494    /// cannot see — that a path is never reused by a different blob.
495    pub fn write_once(mut self) -> Self {
496        self.replacement = crate::blob::BlobReplacement::WriteOnce;
497        self
498    }
499
500    /// Take the blob id from `column` instead of the primary key.
501    pub fn with_id_column(mut self, column: impl Into<String>) -> Self {
502        self.id_column = column.into();
503        self
504    }
505
506    /// Take the plaintext byte length from `column` instead of `size`.
507    pub fn with_size_column(mut self, column: impl Into<String>) -> Self {
508        self.size_column = column.into();
509        self
510    }
511
512    /// Take the content hash from `column` instead of `hash`.
513    pub fn with_hash_column(mut self, column: impl Into<String>) -> Self {
514        self.hash_column = column.into();
515        self
516    }
517
518    /// Key the blob at the readable cloud path in `column` (the plain scheme).
519    pub fn with_cloud_path_column(mut self, column: impl Into<String>) -> Self {
520        self.cloud_path_column = Some(column.into());
521        self
522    }
523
524    /// Scope the blob's encryption (defaults to [`crate::blob::BlobScope::Master`]).
525    pub fn with_scope(mut self, scope: crate::blob::BlobScope) -> Self {
526        self.scope = scope;
527        self
528    }
529}
530
531/// Column names of `table`, in declared order, via `PRAGMA table_info`. The
532/// index of a name here is the index SQLite session changesets report for that
533/// column.
534pub(crate) fn table_columns(conn: &Connection, table: &str) -> rusqlite::Result<Vec<String>> {
535    let sql = format!("PRAGMA table_info({})", quote_ident(table));
536    let mut stmt = conn.prepare(&sql)?;
537    let columns = stmt
538        .query_map([], |row| row.get::<_, String>(1))?
539        .collect::<Result<Vec<_>, _>>()?;
540    Ok(columns)
541}
542
543/// Quote an SQL identifier (table/column name), doubling any embedded quote, so
544/// a trusted-but-unbindable name interpolates safely. Identifiers cannot be
545/// passed as bound parameters; this is the safe interpolation path for them.
546pub(crate) fn quote_ident(ident: &str) -> String {
547    format!("\"{}\"", ident.replace('"', "\"\""))
548}
549
550#[derive(Debug, thiserror::Error)]
551pub(crate) enum ForeignKeySchemaError {
552    #[error(transparent)]
553    Sqlite(#[from] rusqlite::Error),
554    #[error("foreign key on {child_table:?} is malformed: {reason}")]
555    Malformed { child_table: String, reason: String },
556}
557
558#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord)]
559pub(crate) struct ForeignKeyColumn {
560    pub(crate) child: String,
561    pub(crate) parent: String,
562}
563
564#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord)]
565pub(crate) struct ForeignKeyEdge {
566    pub(crate) parent_table: String,
567    pub(crate) columns: Vec<ForeignKeyColumn>,
568    pub(crate) on_update: String,
569    pub(crate) on_delete: String,
570    pub(crate) match_clause: String,
571}
572
573struct ForeignKeyRow {
574    sequence: i64,
575    parent_table: String,
576    child_column: String,
577    parent_column: Option<String>,
578    on_update: String,
579    on_delete: String,
580    match_clause: String,
581}
582
583/// Every outgoing foreign key on `child_table`, grouped by constraint and
584/// sorted by its complete parent/column/action shape. SQLite's constraint ids
585/// and PRAGMA listing order are deliberately discarded.
586pub(crate) fn foreign_key_edges(
587    conn: &Connection,
588    child_table: &str,
589) -> Result<Vec<ForeignKeyEdge>, ForeignKeySchemaError> {
590    let sql = format!("PRAGMA foreign_key_list({})", quote_ident(child_table));
591    let mut statement = conn.prepare(&sql)?;
592    let rows = statement.query_map([], |row| {
593        Ok((
594            row.get::<_, i64>(0)?,
595            ForeignKeyRow {
596                sequence: row.get(1)?,
597                parent_table: row.get(2)?,
598                child_column: row.get(3)?,
599                parent_column: row.get(4)?,
600                on_update: row.get::<_, String>(5)?.to_ascii_uppercase(),
601                on_delete: row.get::<_, String>(6)?.to_ascii_uppercase(),
602                match_clause: row.get::<_, String>(7)?.to_ascii_uppercase(),
603            },
604        ))
605    })?;
606    let mut grouped: BTreeMap<i64, Vec<ForeignKeyRow>> = BTreeMap::new();
607    for row in rows {
608        let (id, row) = row?;
609        grouped.entry(id).or_default().push(row);
610    }
611
612    let mut edges = Vec::with_capacity(grouped.len());
613    for mut rows in grouped.into_values() {
614        rows.sort_by_key(|row| row.sequence);
615        let first = rows
616            .first()
617            .ok_or_else(|| ForeignKeySchemaError::Malformed {
618                child_table: child_table.to_string(),
619                reason: "constraint has no columns".to_string(),
620            })?;
621        if rows.iter().any(|row| {
622            row.parent_table != first.parent_table
623                || row.on_update != first.on_update
624                || row.on_delete != first.on_delete
625                || row.match_clause != first.match_clause
626        }) {
627            return Err(ForeignKeySchemaError::Malformed {
628                child_table: child_table.to_string(),
629                reason: "one constraint reports inconsistent parent or actions".to_string(),
630            });
631        }
632        let omitted_parent_columns = rows.iter().all(|row| row.parent_column.is_none());
633        if !omitted_parent_columns && rows.iter().any(|row| row.parent_column.is_none()) {
634            return Err(ForeignKeySchemaError::Malformed {
635                child_table: child_table.to_string(),
636                reason: "one constraint mixes named and omitted parent columns".to_string(),
637            });
638        }
639        let inferred_parent_columns = if omitted_parent_columns {
640            primary_key_columns(conn, &first.parent_table)?
641        } else {
642            Vec::new()
643        };
644        if omitted_parent_columns && inferred_parent_columns.len() != rows.len() {
645            return Err(ForeignKeySchemaError::Malformed {
646                child_table: child_table.to_string(),
647                reason: format!(
648                    "{} child columns reference {} primary-key columns",
649                    rows.len(),
650                    inferred_parent_columns.len(),
651                ),
652            });
653        }
654        let columns = rows
655            .iter()
656            .enumerate()
657            .map(|(position, row)| ForeignKeyColumn {
658                child: row.child_column.clone(),
659                parent: row
660                    .parent_column
661                    .clone()
662                    .unwrap_or_else(|| inferred_parent_columns[position].clone()),
663            })
664            .collect();
665        edges.push(ForeignKeyEdge {
666            parent_table: first.parent_table.clone(),
667            columns,
668            on_update: first.on_update.clone(),
669            on_delete: first.on_delete.clone(),
670            match_clause: first.match_clause.clone(),
671        });
672    }
673    edges.sort();
674    Ok(edges)
675}
676
677fn primary_key_columns(
678    conn: &Connection,
679    table: &str,
680) -> Result<Vec<String>, ForeignKeySchemaError> {
681    let sql = format!("PRAGMA table_info({})", quote_ident(table));
682    let mut statement = conn.prepare(&sql)?;
683    let rows = statement.query_map([], |row| {
684        Ok((row.get::<_, i64>(5)?, row.get::<_, String>(1)?))
685    })?;
686    let mut columns = rows
687        .collect::<rusqlite::Result<Vec<_>>>()?
688        .into_iter()
689        .filter(|(rank, _)| *rank > 0)
690        .collect::<Vec<_>>();
691    columns.sort_by_key(|(rank, _)| *rank);
692    Ok(columns.into_iter().map(|(_, name)| name).collect())
693}
694
695#[cfg(test)]
696mod row_identity_tests {
697    use super::*;
698
699    #[test]
700    fn independent_uuid_accepts_only_canonical_rfc_uuid_v4_or_v7() {
701        for valid in [
702            "f47ac10b-58cc-4372-a567-0e02b2c3d479",
703            "01890a5d-ac96-774b-bcce-b302099c3f74",
704        ] {
705            validate_row_identity("things", RowIdentity::IndependentUuid, valid)
706                .unwrap_or_else(|error| panic!("{valid} must be accepted: {error}"));
707        }
708
709        for invalid in [
710            "F47AC10B-58CC-4372-A567-0E02B2C3D479",
711            "f47ac10b58cc4372a5670e02b2c3d479",
712            "{f47ac10b-58cc-4372-a567-0e02b2c3d479}",
713            "urn:uuid:f47ac10b-58cc-4372-a567-0e02b2c3d479",
714            "00000000-0000-0000-0000-000000000000",
715            "f47ac10b-58cc-1372-a567-0e02b2c3d479",
716            "f47ac10b-58cc-2372-a567-0e02b2c3d479",
717            "f47ac10b-58cc-3372-a567-0e02b2c3d479",
718            "f47ac10b-58cc-5372-a567-0e02b2c3d479",
719            "f47ac10b-58cc-6372-a567-0e02b2c3d479",
720            "f47ac10b-58cc-8372-a567-0e02b2c3d479",
721            "f47ac10b-58cc-4372-0567-0e02b2c3d479",
722            "not-a-uuid",
723        ] {
724            assert!(
725                validate_row_identity("things", RowIdentity::IndependentUuid, invalid).is_err(),
726                "{invalid} must be rejected",
727            );
728        }
729
730        validate_row_identity("settings", RowIdentity::SharedKey, "preferences")
731            .expect("shared keys accept application-assigned ids");
732    }
733}
coven_core/sync/session.rs

coven_core/sync/
session.rs
