coven_database/
changeset_identity.rs

1use fallible_streaming_iterator::FallibleStreamingIterator;
2use rusqlite::hooks::Action;
3use rusqlite::session::{ChangesetItem, ChangesetIter};
4use rusqlite::types::ValueRef;
5
6use coven_protocol::synced_schema::{RowIdentityError, SyncedTable};
7
8#[derive(Debug, thiserror::Error)]
9pub enum ChangesetIdentityError {
10    #[error("changeset row identity validation failed: {0}")]
11    Parse(#[from] rusqlite::Error),
12    #[error("changeset contains undeclared table {0:?}")]
13    UndeclaredTable(String),
14    #[error(transparent)]
15    Row(#[from] RowIdentityError),
16}
17
18pub(crate) fn validate_changeset_row_identities(
19    bytes: &[u8],
20    tables: &[SyncedTable],
21) -> Result<(), ChangesetIdentityError> {
22    if bytes.is_empty() {
23        return Ok(());
24    }
25
26    let input: &mut dyn std::io::Read = &mut &bytes[..];
27    let mut iter = ChangesetIter::start_strm(&input).map_err(ChangesetIdentityError::Parse)?;
28    while let Some(item) = iter.next().map_err(ChangesetIdentityError::Parse)? {
29        let op = item.op().map_err(ChangesetIdentityError::Parse)?;
30        let table_name = op.table_name();
31        let table = tables
32            .iter()
33            .find(|table| table.name() == table_name)
34            .ok_or_else(|| ChangesetIdentityError::UndeclaredTable(table_name.to_string()))?;
35        match op.code() {
36            Action::SQLITE_INSERT => {
37                let id = required_changeset_id(item, table_name, "new", ChangesetSide::New)?;
38                table.row_identity().validate(table_name, &id)?;
39            }
40            Action::SQLITE_DELETE => {
41                let id = required_changeset_id(item, table_name, "old", ChangesetSide::Old)?;
42                table.row_identity().validate(table_name, &id)?;
43            }
44            Action::SQLITE_UPDATE => {
45                let old = required_changeset_id(item, table_name, "old", ChangesetSide::Old)?;
46                let id = optional_changeset_id(item, table_name, "new", ChangesetSide::New)?
47                    .unwrap_or(old);
48                table.row_identity().validate(table_name, &id)?;
49            }
50            _ => {}
51        }
52    }
53    Ok(())
54}
55
56#[derive(Clone, Copy)]
57enum ChangesetSide {
58    Old,
59    New,
60}
61
62fn required_changeset_id(
63    item: &ChangesetItem,
64    table: &str,
65    side_name: &'static str,
66    side: ChangesetSide,
67) -> Result<String, ChangesetIdentityError> {
68    optional_changeset_id(item, table, side_name, side)?.ok_or_else(|| {
69        ChangesetIdentityError::Row(RowIdentityError::MissingPrimaryKey {
70            table: table.to_string(),
71            side: side_name,
72        })
73    })
74}
75
76fn optional_changeset_id(
77    item: &ChangesetItem,
78    table: &str,
79    side_name: &'static str,
80    side: ChangesetSide,
81) -> Result<Option<String>, ChangesetIdentityError> {
82    let value = match side {
83        ChangesetSide::Old => item.old_value(0),
84        ChangesetSide::New => item.new_value(0),
85    };
86    let value = match value {
87        Ok(value) => value,
88        Err(rusqlite::Error::InvalidColumnIndex(_)) => return Ok(None),
89        Err(error) => return Err(ChangesetIdentityError::Parse(error)),
90    };
91    let ValueRef::Text(bytes) = value else {
92        return Err(RowIdentityError::NonTextPrimaryKey {
93            table: table.to_string(),
94            side: side_name,
95        }
96        .into());
97    };
98    std::str::from_utf8(bytes)
99        .map(str::to_owned)
100        .map(Some)
101        .map_err(|error| RowIdentityError::NonUtf8PrimaryKey {
102            table: table.to_string(),
103            source: error,
104        })
105        .map_err(ChangesetIdentityError::from)
106}
coven_database/changeset_identity.rs

coven_database/
changeset_identity.rs
