1use std::{
2 fs,
3 path::{Path, PathBuf},
4 sync::Mutex,
5 time::Duration,
6};
7
8use kcode_k1_access_profile_wire::{decode_profile, encode_profile};
9use rusqlite::{
10 Connection, ErrorCode, OpenFlags, OptionalExtension, Row, TransactionBehavior, params,
11};
12
13pub use kcode_k1_access_profile_values::{
14 AuthorizationProfile, ProfileId, ProfileRevision, TxId, UserId,
15};
16
17const DATABASE_FILE: &str = "profiles.sqlite3";
18const METADATA_SQL: &str = "CREATE TABLE metadata(schema_version INTEGER NOT NULL CHECK(schema_version=1), last_applied_txid BLOB CHECK(last_applied_txid IS NULL OR (typeof(last_applied_txid)='blob' AND length(last_applied_txid)=12)))";
19const PROFILES_SQL: &str = "CREATE TABLE profiles(profile_id BLOB PRIMARY KEY NOT NULL CHECK(typeof(profile_id)='blob' AND length(profile_id)=12), owner BLOB NOT NULL CHECK(typeof(owner)='blob' AND length(owner)=12), revision BLOB NOT NULL CHECK(typeof(revision)='blob' AND length(revision)=12), profile BLOB NOT NULL CHECK(typeof(profile)='blob')) WITHOUT ROWID";
20const OWNER_INDEX_SQL: &str = "CREATE INDEX profiles_owner ON profiles(owner)";
21const CREATE_SCHEMA: &str = "BEGIN IMMEDIATE;
22CREATE TABLE metadata(schema_version INTEGER NOT NULL CHECK(schema_version=1), last_applied_txid BLOB CHECK(last_applied_txid IS NULL OR (typeof(last_applied_txid)='blob' AND length(last_applied_txid)=12)));
23CREATE TABLE profiles(profile_id BLOB PRIMARY KEY NOT NULL CHECK(typeof(profile_id)='blob' AND length(profile_id)=12), owner BLOB NOT NULL CHECK(typeof(owner)='blob' AND length(owner)=12), revision BLOB NOT NULL CHECK(typeof(revision)='blob' AND length(revision)=12), profile BLOB NOT NULL CHECK(typeof(profile)='blob')) WITHOUT ROWID;
24CREATE INDEX profiles_owner ON profiles(owner);
25INSERT INTO metadata(schema_version, last_applied_txid) VALUES(1, NULL);
26COMMIT;";
27
28#[derive(Clone, Debug, Eq, PartialEq)]
29pub struct SavedProfile {
30 profile_id: ProfileId,
31 owner: UserId,
32 revision: ProfileRevision,
33 profile: AuthorizationProfile,
34}
35
36impl SavedProfile {
37 pub const fn profile_id(&self) -> ProfileId {
38 self.profile_id
39 }
40
41 pub const fn owner(&self) -> UserId {
42 self.owner
43 }
44
45 pub const fn revision(&self) -> ProfileRevision {
46 self.revision
47 }
48
49 pub fn profile(&self) -> &AuthorizationProfile {
50 &self.profile
51 }
52}
53
54#[derive(Clone, Debug, Eq, PartialEq)]
55pub enum ProfileAction {
56 Create {
57 owner: UserId,
58 profile: AuthorizationProfile,
59 },
60 Replace {
61 profile_id: ProfileId,
62 actor: UserId,
63 profile: AuthorizationProfile,
64 },
65 Delete {
66 profile_id: ProfileId,
67 actor: UserId,
68 },
69}
70
71#[derive(Clone, Debug, Eq, PartialEq)]
72pub enum ApplyOutcome {
73 Applied(ProfileRevision),
74 Unchanged(ProfileRevision),
75 Rejected(String),
76}
77
78#[derive(Clone, Debug, Eq, PartialEq)]
79pub struct Snapshot {
80 cursor: Option<TxId>,
81 profiles: Vec<SavedProfile>,
82}
83
84impl Snapshot {
85 pub const fn cursor(&self) -> Option<TxId> {
86 self.cursor
87 }
88
89 pub fn profiles(&self) -> &[SavedProfile] {
90 &self.profiles
91 }
92}
93
94#[derive(Clone, Debug, Eq, PartialEq)]
95pub enum OpenError {
96 Rebuildable,
97 Fatal(String),
98}
99
100pub struct ProfileDatabase {
101 database: PathBuf,
102 apply_connection: Mutex<Connection>,
103}
104
105impl ProfileDatabase {
106 pub fn open(root: &Path) -> Result<(Self, Snapshot), OpenError> {
107 fs::create_dir_all(root)
108 .map_err(|error| OpenError::Fatal(format!("profile root is unavailable: {error}")))?;
109 let database = root.join(DATABASE_FILE);
110 let (connection, snapshot) = load_database(&database)?;
111 Ok((
112 Self {
113 database,
114 apply_connection: Mutex::new(connection),
115 },
116 snapshot,
117 ))
118 }
119
120 pub fn rebuild(root: &Path) -> Result<(Self, Snapshot), String> {
121 remove_database(root)?;
122 match Self::open(root) {
123 Ok(value) => Ok(value),
124 Err(OpenError::Fatal(error)) => Err(error),
125 Err(OpenError::Rebuildable) => Err("profile database rebuild failed".to_owned()),
126 }
127 }
128
129 pub fn apply(
130 &self,
131 callback_txid: TxId,
132 action: ProfileAction,
133 ) -> Result<ApplyOutcome, String> {
134 let encoded = match &action {
135 ProfileAction::Create { profile, .. } | ProfileAction::Replace { profile, .. } => {
136 Some(encode_profile(profile)?)
137 }
138 ProfileAction::Delete { .. } => None,
139 };
140 let mut connection = self
141 .apply_connection
142 .lock()
143 .map_err(|_| "profile apply lane is unavailable".to_owned())?;
144 let transaction = connection
145 .transaction_with_behavior(TransactionBehavior::Immediate)
146 .map_err(persistence_error)?;
147 let outcome = apply_action(&transaction, callback_txid, action, encoded)?;
148 transaction.commit().map_err(persistence_error)?;
149 Ok(outcome)
150 }
151
152 pub fn get_for_user(
153 &self,
154 user: UserId,
155 profile_id: ProfileId,
156 ) -> Result<Option<SavedProfile>, String> {
157 query_profile(&self.database, user, profile_id)
158 }
159
160 pub fn list_for_user(&self, user: UserId) -> Result<Vec<SavedProfile>, String> {
161 query_profiles(&self.database, user)
162 }
163
164 pub fn clear(&self) -> Result<(), String> {
165 let mut connection = self
166 .apply_connection
167 .lock()
168 .map_err(|_| "profile apply lane is unavailable".to_owned())?;
169 let transaction = connection
170 .transaction_with_behavior(TransactionBehavior::Immediate)
171 .map_err(persistence_error)?;
172 transaction
173 .execute("DELETE FROM profiles", [])
174 .map_err(persistence_error)?;
175 changed_one(transaction.execute("UPDATE metadata SET last_applied_txid=NULL", []))?;
176 transaction.commit().map_err(persistence_error)
177 }
178}
179
180fn apply_action(
181 tx: &rusqlite::Transaction<'_>,
182 callback: TxId,
183 action: ProfileAction,
184 encoded: Option<Vec<u8>>,
185) -> Result<ApplyOutcome, String> {
186 let outcome = match action {
187 ProfileAction::Create { owner, .. } => {
188 let id = ProfileId::new(callback);
189 if load_saved(tx, id)?.is_some() {
190 ApplyOutcome::Rejected("profile already exists".to_owned())
191 } else {
192 let profile = encoded
193 .as_deref()
194 .ok_or("profile encoding is unavailable")?;
195 let id_bytes = callback.into_bytes();
196 let owner_bytes = owner.as_tx_id().into_bytes();
197 changed_one(tx.execute(
198 "INSERT INTO profiles(profile_id, owner, revision, profile) VALUES(?1, ?2, ?3, ?4)",
199 params![
200 id_bytes.as_slice(),
201 owner_bytes.as_slice(),
202 id_bytes.as_slice(),
203 profile
204 ],
205 ))?;
206 ApplyOutcome::Applied(ProfileRevision::new(id, callback))
207 }
208 }
209 ProfileAction::Replace {
210 profile_id,
211 actor,
212 profile,
213 } => match load_saved(tx, profile_id)? {
214 Some(current) if current.owner() == actor => {
215 if current.profile() == &profile {
216 ApplyOutcome::Unchanged(current.revision())
217 } else {
218 let blob = encoded
219 .as_deref()
220 .ok_or("profile encoding is unavailable")?;
221 let id = profile_id.txid().into_bytes();
222 let revision = callback.into_bytes();
223 changed_one(tx.execute(
224 "UPDATE profiles SET revision=?1, profile=?2 WHERE profile_id=?3",
225 params![revision.as_slice(), blob, id.as_slice()],
226 ))?;
227 ApplyOutcome::Applied(ProfileRevision::new(profile_id, callback))
228 }
229 }
230 _ => ApplyOutcome::Rejected("profile is unavailable".to_owned()),
231 },
232 ProfileAction::Delete { profile_id, actor } => match load_saved(tx, profile_id)? {
233 Some(current) if current.owner() == actor => {
234 let id = profile_id.txid().into_bytes();
235 changed_one(tx.execute(
236 "DELETE FROM profiles WHERE profile_id=?1",
237 params![id.as_slice()],
238 ))?;
239 ApplyOutcome::Applied(ProfileRevision::new(profile_id, callback))
240 }
241 _ => ApplyOutcome::Rejected("profile is unavailable".to_owned()),
242 },
243 };
244 let cursor = callback.into_bytes();
245 changed_one(tx.execute(
246 "UPDATE metadata SET last_applied_txid=?1",
247 params![cursor.as_slice()],
248 ))?;
249 Ok(outcome)
250}
251
252fn changed_one(result: rusqlite::Result<usize>) -> Result<(), String> {
253 match result.map_err(persistence_error)? {
254 1 => Ok(()),
255 _ => Err("profile persistence contradiction".to_owned()),
256 }
257}
258
259fn persistence_error(error: rusqlite::Error) -> String {
260 format!("profile persistence failed: {error}")
261}
262
263fn load_saved(
264 tx: &rusqlite::Transaction<'_>,
265 id: ProfileId,
266) -> Result<Option<SavedProfile>, String> {
267 let bytes = id.txid().into_bytes();
268 let raw = tx
269 .query_row(
270 "SELECT profile_id, owner, revision, profile FROM profiles WHERE profile_id=?1",
271 params![bytes.as_slice()],
272 raw_profile,
273 )
274 .optional()
275 .map_err(persistence_error)?;
276 raw.map(decode_saved).transpose()
277}
278
279fn query_profile(
280 database: &Path,
281 user: UserId,
282 id: ProfileId,
283) -> Result<Option<SavedProfile>, String> {
284 let connection = open_query(database)?;
285 let owner = user.as_tx_id().into_bytes();
286 let profile = id.txid().into_bytes();
287 let raw = connection
288 .query_row(
289 "SELECT profile_id, owner, revision, profile FROM profiles WHERE profile_id=?1 AND owner=?2",
290 params![profile.as_slice(), owner.as_slice()],
291 raw_profile,
292 )
293 .optional()
294 .map_err(query_error)?;
295 raw.map(decode_saved).transpose()
296}
297
298fn query_profiles(database: &Path, user: UserId) -> Result<Vec<SavedProfile>, String> {
299 let connection = open_query(database)?;
300 let owner = user.as_tx_id().into_bytes();
301 let mut statement = connection
302 .prepare("SELECT profile_id, owner, revision, profile FROM profiles WHERE owner=?1")
303 .map_err(query_error)?;
304 let mut rows = statement
305 .query(params![owner.as_slice()])
306 .map_err(query_error)?;
307 let mut profiles = Vec::new();
308 while let Some(row) = rows.next().map_err(query_error)? {
309 profiles.push(decode_saved(raw_profile(row).map_err(query_error)?)?);
310 }
311 Ok(profiles)
312}
313
314fn open_query(database: &Path) -> Result<Connection, String> {
315 let flags = OpenFlags::SQLITE_OPEN_READ_ONLY | OpenFlags::SQLITE_OPEN_NO_MUTEX;
316 let connection = Connection::open_with_flags(database, flags).map_err(query_error)?;
317 connection
318 .busy_timeout(Duration::ZERO)
319 .map_err(query_error)?;
320 Ok(connection)
321}
322
323fn query_error(error: rusqlite::Error) -> String {
324 format!("profile query failed: {error}")
325}
326
327struct RawProfile {
328 profile_id: Vec<u8>,
329 owner: Vec<u8>,
330 revision: Vec<u8>,
331 profile: Vec<u8>,
332}
333
334fn raw_profile(row: &Row<'_>) -> rusqlite::Result<RawProfile> {
335 Ok(RawProfile {
336 profile_id: row.get(0)?,
337 owner: row.get(1)?,
338 revision: row.get(2)?,
339 profile: row.get(3)?,
340 })
341}
342
343fn decode_saved(raw: RawProfile) -> Result<SavedProfile, String> {
344 let profile_id = ProfileId::new(TxId::from_bytes(bytes_12(raw.profile_id)?));
345 let owner = UserId::from_tx_id(TxId::from_bytes(bytes_12(raw.owner)?));
346 let revision = TxId::from_bytes(bytes_12(raw.revision)?);
347 let profile = decode_profile(&raw.profile).map_err(|_| "profile persistence contradiction")?;
348 if encode_profile(&profile).map_err(|_| "profile persistence contradiction")? != raw.profile {
349 return Err("profile persistence contradiction".to_owned());
350 }
351 Ok(SavedProfile {
352 profile_id,
353 owner,
354 revision: ProfileRevision::new(profile_id, revision),
355 profile,
356 })
357}
358
359fn bytes_12(bytes: Vec<u8>) -> Result<[u8; 12], String> {
360 bytes
361 .try_into()
362 .map_err(|_| "profile persistence contradiction".to_owned())
363}
364
365fn load_database(database: &Path) -> Result<(Connection, Snapshot), OpenError> {
366 let connection = connect(database)?;
367 let count: i64 = connection
368 .query_row(
369 "SELECT count(*) FROM sqlite_schema WHERE name NOT LIKE 'sqlite_%'",
370 [],
371 |row| row.get(0),
372 )
373 .map_err(sql_issue)?;
374 if count == 0 {
375 connection.execute_batch(CREATE_SCHEMA).map_err(sql_issue)?;
376 }
377 validate_schema(&connection)?;
378 let snapshot = validate_rows(&connection)?;
379 Ok((connection, snapshot))
380}
381
382fn connect(database: &Path) -> Result<Connection, OpenError> {
383 let connection = Connection::open(database).map_err(sql_issue)?;
384 connection.busy_timeout(Duration::ZERO).map_err(sql_issue)?;
385 let journal: String = connection
386 .query_row("PRAGMA journal_mode=WAL", [], |row| row.get(0))
387 .map_err(sql_issue)?;
388 if !journal.eq_ignore_ascii_case("wal") {
389 return Err(OpenError::Fatal("SQLite WAL is unavailable".to_owned()));
390 }
391 connection
392 .execute_batch("PRAGMA synchronous=FULL;")
393 .map_err(sql_issue)?;
394 let synchronous: i64 = connection
395 .query_row("PRAGMA synchronous", [], |row| row.get(0))
396 .map_err(sql_issue)?;
397 if synchronous != 2 {
398 return Err(OpenError::Fatal(
399 "SQLite FULL synchronization is unavailable".to_owned(),
400 ));
401 }
402 Ok(connection)
403}
404
405fn validate_schema(connection: &Connection) -> Result<(), OpenError> {
406 let integrity: String = connection
407 .query_row("PRAGMA integrity_check", [], |row| row.get(0))
408 .map_err(sql_issue)?;
409 if integrity != "ok" {
410 return Err(OpenError::Rebuildable);
411 }
412 let mut statement = connection
413 .prepare("SELECT name, sql FROM sqlite_schema WHERE name NOT LIKE 'sqlite_%'")
414 .map_err(sql_issue)?;
415 let mut rows = statement.query([]).map_err(sql_issue)?;
416 let mut found = 0;
417 while let Some(row) = rows.next().map_err(sql_issue)? {
418 let name: String = row.get(0).map_err(|_| OpenError::Rebuildable)?;
419 let sql: String = row.get(1).map_err(|_| OpenError::Rebuildable)?;
420 match (name.as_str(), sql.as_str()) {
421 ("metadata", METADATA_SQL)
422 | ("profiles", PROFILES_SQL)
423 | ("profiles_owner", OWNER_INDEX_SQL) => found += 1,
424 _ => return Err(OpenError::Rebuildable),
425 }
426 }
427 (found == 3).then_some(()).ok_or(OpenError::Rebuildable)
428}
429
430fn validate_rows(connection: &Connection) -> Result<Snapshot, OpenError> {
431 let count: i64 = connection
432 .query_row("SELECT count(*) FROM metadata", [], |row| row.get(0))
433 .map_err(sql_issue)?;
434 if count != 1 {
435 return Err(OpenError::Rebuildable);
436 }
437 let (version, cursor): (i64, Option<Vec<u8>>) = connection
438 .query_row(
439 "SELECT schema_version, last_applied_txid FROM metadata",
440 [],
441 |row| Ok((row.get(0)?, row.get(1)?)),
442 )
443 .map_err(|_| OpenError::Rebuildable)?;
444 if version != 1 {
445 return Err(OpenError::Rebuildable);
446 }
447 let cursor = cursor
448 .map(|bytes| bytes_12(bytes).map(TxId::from_bytes))
449 .transpose()
450 .map_err(|_| OpenError::Rebuildable)?;
451 let mut statement = connection
452 .prepare("SELECT profile_id, owner, revision, profile FROM profiles")
453 .map_err(sql_issue)?;
454 let mut rows = statement.query([]).map_err(sql_issue)?;
455 let mut profiles = Vec::new();
456 while let Some(row) = rows.next().map_err(sql_issue)? {
457 let raw = raw_profile(row).map_err(|_| OpenError::Rebuildable)?;
458 profiles.push(decode_saved(raw).map_err(|_| OpenError::Rebuildable)?);
459 }
460 if !profiles.is_empty() && cursor.is_none() {
461 return Err(OpenError::Rebuildable);
462 }
463 Ok(Snapshot { cursor, profiles })
464}
465
466fn sql_issue(error: rusqlite::Error) -> OpenError {
467 match error.sqlite_error_code() {
468 Some(ErrorCode::DatabaseCorrupt | ErrorCode::NotADatabase) => OpenError::Rebuildable,
469 _ => OpenError::Fatal(format!("SQLite is unavailable: {error}")),
470 }
471}
472
473fn remove_database(root: &Path) -> Result<(), String> {
474 for file in [
475 "profiles.sqlite3-wal",
476 "profiles.sqlite3-shm",
477 DATABASE_FILE,
478 ] {
479 match fs::remove_file(root.join(file)) {
480 Ok(()) => (),
481 Err(error) if error.kind() == std::io::ErrorKind::NotFound => (),
482 Err(error) => return Err(format!("profile database cannot be rebuilt: {error}")),
483 }
484 }
485 Ok(())
486}
487
488#[cfg(test)]
489mod tests {
490 use super::*;
491 use std::time::{SystemTime, UNIX_EPOCH};
492
493 #[test]
494 fn rejected_delete_cursor_survives_reopen_and_clear_resets_it() {
495 let nonce = SystemTime::now()
496 .duration_since(UNIX_EPOCH)
497 .unwrap()
498 .as_nanos();
499 let root = std::env::temp_dir().join(format!("profile-sqlite-test-{nonce}"));
500 let tx = |byte| TxId::from_bytes([byte; 12]);
501 let (database, _) = ProfileDatabase::open(&root).unwrap();
502 let action = ProfileAction::Delete {
503 profile_id: ProfileId::new(tx(1)),
504 actor: UserId::from_tx_id(tx(2)),
505 };
506 let outcome = database.apply(tx(3), action).unwrap();
507 let rejected = ApplyOutcome::Rejected("profile is unavailable".to_owned());
508 assert_eq!(outcome, rejected);
509 drop(database);
510
511 let (database, snapshot) = ProfileDatabase::open(&root).unwrap();
512 assert_eq!(snapshot.cursor(), Some(tx(3)));
513 assert!(snapshot.profiles().is_empty());
514 database.clear().unwrap();
515 drop(database);
516
517 let (database, snapshot) = ProfileDatabase::open(&root).unwrap();
518 assert_eq!(snapshot.cursor(), None);
519 assert!(snapshot.profiles().is_empty());
520 drop(database);
521 fs::remove_dir_all(root).unwrap();
522 }
523}