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