kcode_k1_access_profile_store/
lib.rs

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

kcode_k1_access_profile_store/
lib.rs
