1use crate::{SqliteSyncError, SqliteSyncResult, SqliteSyncStore};
14use rusqlite::{params, OptionalExtension, TransactionBehavior};
15
16const MAX_NAMESPACE_BYTES: usize = 128;
17const MAX_OPAQUE_KEY_BYTES: usize = 512;
18
19#[derive(Debug, Clone, PartialEq, Eq)]
21pub struct SqliteSyncTombstone {
22 pub namespace: String,
24 pub opaque_key: String,
26 pub deleted_sequence: u64,
28 pub payload_hash: String,
30 pub expires_at_ms: u64,
32}
33
34#[derive(Debug, Clone)]
36pub struct SqliteSyncTombstoneStore {
37 store: SqliteSyncStore,
38}
39
40impl SqliteSyncTombstoneStore {
41 pub(crate) fn new(store: SqliteSyncStore) -> Self {
42 Self { store }
43 }
44
45 pub fn record(&self, tombstone: &SqliteSyncTombstone) -> SqliteSyncResult<bool> {
47 validate_tombstone(tombstone)?;
48 self.store.with_connection(|connection| {
49 let transaction = connection
50 .transaction_with_behavior(TransactionBehavior::Immediate)
51 .map_err(SqliteSyncError::database)?;
52 let existing: Option<(i64, String, i64)> = transaction
53 .query_row(
54 "SELECT deleted_sequence, payload_hash, expires_at_ms
55 FROM appcore_sync_tombstone WHERE namespace = ?1 AND opaque_key = ?2",
56 params![tombstone.namespace, tombstone.opaque_key],
57 |row| Ok((row.get(0)?, row.get(1)?, row.get(2)?)),
58 )
59 .optional()
60 .map_err(SqliteSyncError::database)?;
61 let incoming_sequence = to_i64(tombstone.deleted_sequence, "tombstone sequence")?;
62 let incoming_expiry = to_i64(tombstone.expires_at_ms, "tombstone expiry")?;
63 if let Some((sequence, hash, expiry)) = existing {
64 if incoming_sequence < sequence {
65 return Ok(false);
66 }
67 if incoming_sequence == sequence {
68 return if hash == tombstone.payload_hash && expiry == incoming_expiry {
69 Ok(false)
70 } else {
71 Err(SqliteSyncError::CorruptRecord("tombstone conflict"))
72 };
73 }
74 } else if count_tombstones(&transaction)? >= self.store.config().max_tombstones {
75 return Err(SqliteSyncError::CapacityExceeded("tombstone"));
76 }
77 let changed = transaction
78 .execute(
79 "INSERT INTO appcore_sync_tombstone
80 (namespace, opaque_key, deleted_sequence, payload_hash, expires_at_ms)
81 VALUES (?1, ?2, ?3, ?4, ?5)
82 ON CONFLICT(namespace, opaque_key) DO UPDATE SET
83 deleted_sequence = excluded.deleted_sequence,
84 payload_hash = excluded.payload_hash,
85 expires_at_ms = excluded.expires_at_ms
86 WHERE excluded.deleted_sequence >= appcore_sync_tombstone.deleted_sequence",
87 params![
88 tombstone.namespace,
89 tombstone.opaque_key,
90 incoming_sequence,
91 tombstone.payload_hash,
92 incoming_expiry
93 ],
94 )
95 .map_err(SqliteSyncError::database)?;
96 transaction.commit().map_err(SqliteSyncError::database)?;
97 Ok(changed == 1)
98 })
99 }
100
101 pub fn active(&self, now_ms: u64, limit: usize) -> SqliteSyncResult<Vec<SqliteSyncTombstone>> {
103 if limit == 0 || limit > self.store.config().max_tombstones {
104 return Err(SqliteSyncError::CapacityExceeded("tombstone read"));
105 }
106 self.store.with_connection(|connection| {
107 let mut statement = connection
108 .prepare(
109 "SELECT namespace, opaque_key, deleted_sequence, payload_hash, expires_at_ms
110 FROM appcore_sync_tombstone WHERE expires_at_ms > ?1
111 ORDER BY namespace, opaque_key LIMIT ?2",
112 )
113 .map_err(SqliteSyncError::database)?;
114 let rows = statement
115 .query_map(
116 params![
117 to_i64(now_ms, "current time")?,
118 to_i64(limit as u64, "limit")?
119 ],
120 |row| {
121 Ok((
122 row.get::<_, String>(0)?,
123 row.get::<_, String>(1)?,
124 row.get::<_, i64>(2)?,
125 row.get::<_, String>(3)?,
126 row.get::<_, i64>(4)?,
127 ))
128 },
129 )
130 .map_err(SqliteSyncError::database)?;
131 let mut tombstones = Vec::new();
132 for row in rows {
133 let (namespace, opaque_key, sequence, payload_hash, expiry) =
134 row.map_err(SqliteSyncError::database)?;
135 tombstones.push(SqliteSyncTombstone {
136 namespace,
137 opaque_key,
138 deleted_sequence: to_u64(sequence, "tombstone sequence")?,
139 payload_hash,
140 expires_at_ms: to_u64(expiry, "tombstone expiry")?,
141 });
142 }
143 Ok(tombstones)
144 })
145 }
146
147 pub fn prune_expired(&self, now_ms: u64, limit: usize) -> SqliteSyncResult<usize> {
149 if limit == 0 || limit > self.store.config().max_tombstones {
150 return Err(SqliteSyncError::CapacityExceeded("tombstone prune"));
151 }
152 self.store.with_connection(|connection| {
153 connection
154 .execute(
155 "DELETE FROM appcore_sync_tombstone WHERE (namespace, opaque_key) IN (
156 SELECT namespace, opaque_key FROM appcore_sync_tombstone
157 WHERE expires_at_ms <= ?1 ORDER BY expires_at_ms LIMIT ?2
158 )",
159 params![
160 to_i64(now_ms, "current time")?,
161 to_i64(limit as u64, "limit")?
162 ],
163 )
164 .map_err(SqliteSyncError::database)
165 })
166 }
167
168 pub fn len(&self) -> SqliteSyncResult<usize> {
170 self.store
171 .with_connection(|connection| count_tombstones(connection))
172 }
173
174 pub fn is_empty(&self) -> SqliteSyncResult<bool> {
176 self.len().map(|length| length == 0)
177 }
178}
179
180fn count_tombstones(connection: &rusqlite::Connection) -> SqliteSyncResult<usize> {
181 let count: i64 = connection
182 .query_row("SELECT COUNT(*) FROM appcore_sync_tombstone", [], |row| {
183 row.get(0)
184 })
185 .map_err(SqliteSyncError::database)?;
186 usize::try_from(count).map_err(|_| SqliteSyncError::CorruptRecord("tombstone count"))
187}
188
189pub(crate) fn validate_tombstone(tombstone: &SqliteSyncTombstone) -> SqliteSyncResult<()> {
190 if !valid_identifier(&tombstone.namespace, MAX_NAMESPACE_BYTES)
191 || !valid_identifier(&tombstone.opaque_key, MAX_OPAQUE_KEY_BYTES)
192 || tombstone.deleted_sequence == 0
193 || tombstone.expires_at_ms == 0
194 || tombstone.payload_hash.len() != 64
195 || !tombstone
196 .payload_hash
197 .bytes()
198 .all(|byte| byte.is_ascii_hexdigit())
199 {
200 return Err(SqliteSyncError::CorruptRecord("tombstone input"));
201 }
202 Ok(())
203}
204
205fn valid_identifier(value: &str, max_bytes: usize) -> bool {
206 !value.is_empty()
207 && value.len() <= max_bytes
208 && value
209 .chars()
210 .all(|character| character.is_ascii_graphic() && !character.is_ascii_control())
211}
212
213fn to_i64(value: u64, resource: &'static str) -> SqliteSyncResult<i64> {
214 i64::try_from(value).map_err(|_| SqliteSyncError::CapacityExceeded(resource))
215}
216
217fn to_u64(value: i64, resource: &'static str) -> SqliteSyncResult<u64> {
218 u64::try_from(value).map_err(|_| SqliteSyncError::CorruptRecord(resource))
219}