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