1use 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#[derive(Debug, Clone, Copy, PartialEq, Eq)]
25pub enum RowIdentity {
26 IndependentUuid,
29 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#[derive(Debug, Clone, PartialEq, Eq)]
240pub struct SyncedTable {
241 name: String,
242 row_identity: RowIdentity,
243 role: GateRole,
244 blob: Option<BlobDecl>,
245 asset: bool,
250}
251
252#[derive(Debug, Clone, PartialEq, Eq)]
255pub enum GateRole {
256 Plain,
258 RemoteRoot,
261 GatedRoot { gate_column: String },
264 ScopedRoot { audience_column: String },
267 GatedByDescendants,
272}
273
274impl SyncedTable {
275 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 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 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 pub fn remote_root(mut self) -> Self {
310 self.role = GateRole::RemoteRoot;
311 self
312 }
313
314 pub fn gated_by_descendants(mut self) -> Self {
319 self.role = GateRole::GatedByDescendants;
320 self
321 }
322
323 pub fn carries_blob(mut self, decl: BlobDecl) -> Self {
328 self.blob = Some(decl);
329 self
330 }
331
332 pub fn asset(mut self) -> Self {
341 self.asset = true;
342 self
343 }
344
345 pub fn name(&self) -> &str {
347 &self.name
348 }
349
350 pub fn row_identity(&self) -> RowIdentity {
352 self.row_identity
353 }
354
355 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 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 pub fn gate_role(&self) -> &GateRole {
379 &self.role
380 }
381
382 pub fn is_remote_root(&self) -> bool {
385 matches!(self.role, GateRole::RemoteRoot)
386 }
387
388 pub fn is_gated_by_descendants(&self) -> bool {
391 matches!(self.role, GateRole::GatedByDescendants)
392 }
393
394 pub fn blob(&self) -> Option<&BlobDecl> {
396 self.blob.as_ref()
397 }
398
399 pub fn is_asset(&self) -> bool {
402 self.asset
403 }
404}
405
406#[derive(Debug, Clone, PartialEq, Eq)]
413pub struct BlobDecl {
414 pub id_column: String,
417 pub size_column: String,
419 pub hash_column: String,
426 pub namespace: String,
428 pub cloud_path_column: Option<String>,
432 pub scope: crate::blob::BlobScope,
434 pub provenance: crate::blob::Provenance,
438 pub fill: crate::blob::CacheFill,
442 pub replacement: crate::blob::BlobReplacement,
447}
448
449impl BlobDecl {
450 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 pub fn write_once(mut self) -> Self {
480 self.replacement = crate::blob::BlobReplacement::WriteOnce;
481 self
482 }
483
484 pub fn with_id_column(mut self, column: impl Into<String>) -> Self {
486 self.id_column = column.into();
487 self
488 }
489
490 pub fn with_size_column(mut self, column: impl Into<String>) -> Self {
492 self.size_column = column.into();
493 self
494 }
495
496 pub fn with_hash_column(mut self, column: impl Into<String>) -> Self {
498 self.hash_column = column.into();
499 self
500 }
501
502 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 pub fn with_scope(mut self, scope: crate::blob::BlobScope) -> Self {
510 self.scope = scope;
511 self
512 }
513}
514
515pub(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
527pub(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
567pub(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}