use std::{
fs,
path::{Path, PathBuf},
sync::Mutex,
time::Duration,
};
use kcode_k1_access_profile_wire::{decode_profile, encode_profile};
use rusqlite::{
Connection, ErrorCode, OpenFlags, OptionalExtension, Row, TransactionBehavior, params,
};
pub use kcode_k1_access_profile_values::{
AuthorizationProfile, ProfileId, ProfileRevision, TxId, UserId,
};
const DATABASE_FILE: &str = "profiles.sqlite3";
const 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)))";
const 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";
const OWNER_INDEX_SQL: &str = "CREATE INDEX profiles_owner ON profiles(owner)";
const CREATE_SCHEMA: &str = "BEGIN IMMEDIATE;
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)));
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;
CREATE INDEX profiles_owner ON profiles(owner);
INSERT INTO metadata(schema_version, last_applied_txid) VALUES(1, NULL);
COMMIT;";
#[derive(Clone, Debug, Eq, PartialEq)]
pub struct SavedProfile {
profile_id: ProfileId,
owner: UserId,
revision: ProfileRevision,
profile: AuthorizationProfile,
}
impl SavedProfile {
pub const fn profile_id(&self) -> ProfileId {
self.profile_id
}
pub const fn owner(&self) -> UserId {
self.owner
}
pub const fn revision(&self) -> ProfileRevision {
self.revision
}
pub fn profile(&self) -> &AuthorizationProfile {
&self.profile
}
}
#[derive(Clone, Debug, Eq, PartialEq)]
pub enum ProfileAction {
Create {
owner: UserId,
profile: AuthorizationProfile,
},
Replace {
profile_id: ProfileId,
actor: UserId,
profile: AuthorizationProfile,
},
Delete {
profile_id: ProfileId,
actor: UserId,
},
}
#[derive(Clone, Debug, Eq, PartialEq)]
pub enum ApplyOutcome {
Applied(ProfileRevision),
Unchanged(ProfileRevision),
Rejected(String),
}
#[derive(Clone, Debug, Eq, PartialEq)]
pub struct Snapshot {
cursor: Option<TxId>,
profiles: Vec<SavedProfile>,
}
impl Snapshot {
pub const fn cursor(&self) -> Option<TxId> {
self.cursor
}
pub fn profiles(&self) -> &[SavedProfile] {
&self.profiles
}
}
#[derive(Clone, Debug, Eq, PartialEq)]
pub enum OpenError {
Rebuildable,
Fatal(String),
}
pub struct ProfileDatabase {
database: PathBuf,
apply_connection: Mutex<Connection>,
}
impl ProfileDatabase {
pub fn open(root: &Path) -> Result<(Self, Snapshot), OpenError> {
fs::create_dir_all(root)
.map_err(|error| OpenError::Fatal(format!("profile root is unavailable: {error}")))?;
let database = root.join(DATABASE_FILE);
let (connection, snapshot) = load_database(&database)?;
Ok((
Self {
database,
apply_connection: Mutex::new(connection),
},
snapshot,
))
}
pub fn rebuild(root: &Path) -> Result<(Self, Snapshot), String> {
remove_database(root)?;
match Self::open(root) {
Ok(value) => Ok(value),
Err(OpenError::Fatal(error)) => Err(error),
Err(OpenError::Rebuildable) => Err("profile database rebuild failed".to_owned()),
}
}
pub fn apply(
&self,
callback_txid: TxId,
action: ProfileAction,
) -> Result<ApplyOutcome, String> {
let encoded = match &action {
ProfileAction::Create { profile, .. } | ProfileAction::Replace { profile, .. } => {
Some(encode_profile(profile)?)
}
ProfileAction::Delete { .. } => None,
};
let mut connection = self
.apply_connection
.lock()
.map_err(|_| "profile apply lane is unavailable".to_owned())?;
let transaction = connection
.transaction_with_behavior(TransactionBehavior::Immediate)
.map_err(persistence_error)?;
let outcome = apply_action(&transaction, callback_txid, action, encoded)?;
transaction.commit().map_err(persistence_error)?;
Ok(outcome)
}
pub fn get_for_user(
&self,
user: UserId,
profile_id: ProfileId,
) -> Result<Option<SavedProfile>, String> {
query_profile(&self.database, user, profile_id)
}
pub fn list_for_user(&self, user: UserId) -> Result<Vec<SavedProfile>, String> {
query_profiles(&self.database, user)
}
pub fn clear(&self) -> Result<(), String> {
let mut connection = self
.apply_connection
.lock()
.map_err(|_| "profile apply lane is unavailable".to_owned())?;
let transaction = connection
.transaction_with_behavior(TransactionBehavior::Immediate)
.map_err(persistence_error)?;
transaction
.execute("DELETE FROM profiles", [])
.map_err(persistence_error)?;
changed_one(transaction.execute("UPDATE metadata SET last_applied_txid=NULL", []))?;
transaction.commit().map_err(persistence_error)
}
}
fn apply_action(
tx: &rusqlite::Transaction<'_>,
callback: TxId,
action: ProfileAction,
encoded: Option<Vec<u8>>,
) -> Result<ApplyOutcome, String> {
let outcome = match action {
ProfileAction::Create { owner, .. } => {
let id = ProfileId::new(callback);
if load_saved(tx, id)?.is_some() {
ApplyOutcome::Rejected("profile already exists".to_owned())
} else {
let profile = encoded
.as_deref()
.ok_or("profile encoding is unavailable")?;
let id_bytes = callback.into_bytes();
let owner_bytes = owner.as_tx_id().into_bytes();
changed_one(tx.execute(
"INSERT INTO profiles(profile_id, owner, revision, profile) VALUES(?1, ?2, ?3, ?4)",
params![
id_bytes.as_slice(),
owner_bytes.as_slice(),
id_bytes.as_slice(),
profile
],
))?;
ApplyOutcome::Applied(ProfileRevision::new(id, callback))
}
}
ProfileAction::Replace {
profile_id,
actor,
profile,
} => match load_saved(tx, profile_id)? {
Some(current) if current.owner() == actor => {
if current.profile() == &profile {
ApplyOutcome::Unchanged(current.revision())
} else {
let blob = encoded
.as_deref()
.ok_or("profile encoding is unavailable")?;
let id = profile_id.txid().into_bytes();
let revision = callback.into_bytes();
changed_one(tx.execute(
"UPDATE profiles SET revision=?1, profile=?2 WHERE profile_id=?3",
params![revision.as_slice(), blob, id.as_slice()],
))?;
ApplyOutcome::Applied(ProfileRevision::new(profile_id, callback))
}
}
_ => ApplyOutcome::Rejected("profile is unavailable".to_owned()),
},
ProfileAction::Delete { profile_id, actor } => match load_saved(tx, profile_id)? {
Some(current) if current.owner() == actor => {
let id = profile_id.txid().into_bytes();
changed_one(tx.execute(
"DELETE FROM profiles WHERE profile_id=?1",
params![id.as_slice()],
))?;
ApplyOutcome::Applied(ProfileRevision::new(profile_id, callback))
}
_ => ApplyOutcome::Rejected("profile is unavailable".to_owned()),
},
};
let cursor = callback.into_bytes();
changed_one(tx.execute(
"UPDATE metadata SET last_applied_txid=?1",
params![cursor.as_slice()],
))?;
Ok(outcome)
}
fn changed_one(result: rusqlite::Result<usize>) -> Result<(), String> {
match result.map_err(persistence_error)? {
1 => Ok(()),
_ => Err("profile persistence contradiction".to_owned()),
}
}
fn persistence_error(error: rusqlite::Error) -> String {
format!("profile persistence failed: {error}")
}
fn load_saved(
tx: &rusqlite::Transaction<'_>,
id: ProfileId,
) -> Result<Option<SavedProfile>, String> {
let bytes = id.txid().into_bytes();
let raw = tx
.query_row(
"SELECT profile_id, owner, revision, profile FROM profiles WHERE profile_id=?1",
params![bytes.as_slice()],
raw_profile,
)
.optional()
.map_err(persistence_error)?;
raw.map(decode_saved).transpose()
}
fn query_profile(
database: &Path,
user: UserId,
id: ProfileId,
) -> Result<Option<SavedProfile>, String> {
let connection = open_query(database)?;
let owner = user.as_tx_id().into_bytes();
let profile = id.txid().into_bytes();
let raw = connection
.query_row(
"SELECT profile_id, owner, revision, profile FROM profiles WHERE profile_id=?1 AND owner=?2",
params![profile.as_slice(), owner.as_slice()],
raw_profile,
)
.optional()
.map_err(query_error)?;
raw.map(decode_saved).transpose()
}
fn query_profiles(database: &Path, user: UserId) -> Result<Vec<SavedProfile>, String> {
let connection = open_query(database)?;
let owner = user.as_tx_id().into_bytes();
let mut statement = connection
.prepare("SELECT profile_id, owner, revision, profile FROM profiles WHERE owner=?1")
.map_err(query_error)?;
let mut rows = statement
.query(params![owner.as_slice()])
.map_err(query_error)?;
let mut profiles = Vec::new();
while let Some(row) = rows.next().map_err(query_error)? {
profiles.push(decode_saved(raw_profile(row).map_err(query_error)?)?);
}
Ok(profiles)
}
fn open_query(database: &Path) -> Result<Connection, String> {
let flags = OpenFlags::SQLITE_OPEN_READ_ONLY | OpenFlags::SQLITE_OPEN_NO_MUTEX;
let connection = Connection::open_with_flags(database, flags).map_err(query_error)?;
connection
.busy_timeout(Duration::ZERO)
.map_err(query_error)?;
Ok(connection)
}
fn query_error(error: rusqlite::Error) -> String {
format!("profile query failed: {error}")
}
struct RawProfile {
profile_id: Vec<u8>,
owner: Vec<u8>,
revision: Vec<u8>,
profile: Vec<u8>,
}
fn raw_profile(row: &Row<'_>) -> rusqlite::Result<RawProfile> {
Ok(RawProfile {
profile_id: row.get(0)?,
owner: row.get(1)?,
revision: row.get(2)?,
profile: row.get(3)?,
})
}
fn decode_saved(raw: RawProfile) -> Result<SavedProfile, String> {
let profile_id = ProfileId::new(TxId::from_bytes(bytes_12(raw.profile_id)?));
let owner = UserId::from_tx_id(TxId::from_bytes(bytes_12(raw.owner)?));
let revision = TxId::from_bytes(bytes_12(raw.revision)?);
let profile = decode_profile(&raw.profile).map_err(|_| "profile persistence contradiction")?;
if encode_profile(&profile).map_err(|_| "profile persistence contradiction")? != raw.profile {
return Err("profile persistence contradiction".to_owned());
}
Ok(SavedProfile {
profile_id,
owner,
revision: ProfileRevision::new(profile_id, revision),
profile,
})
}
fn bytes_12(bytes: Vec<u8>) -> Result<[u8; 12], String> {
bytes
.try_into()
.map_err(|_| "profile persistence contradiction".to_owned())
}
fn load_database(database: &Path) -> Result<(Connection, Snapshot), OpenError> {
let connection = connect(database)?;
let count: i64 = connection
.query_row(
"SELECT count(*) FROM sqlite_schema WHERE name NOT LIKE 'sqlite_%'",
[],
|row| row.get(0),
)
.map_err(sql_issue)?;
if count == 0 {
connection.execute_batch(CREATE_SCHEMA).map_err(sql_issue)?;
}
validate_schema(&connection)?;
let snapshot = validate_rows(&connection)?;
Ok((connection, snapshot))
}
fn connect(database: &Path) -> Result<Connection, OpenError> {
let connection = Connection::open(database).map_err(sql_issue)?;
connection.busy_timeout(Duration::ZERO).map_err(sql_issue)?;
let journal: String = connection
.query_row("PRAGMA journal_mode=WAL", [], |row| row.get(0))
.map_err(sql_issue)?;
if !journal.eq_ignore_ascii_case("wal") {
return Err(OpenError::Fatal("SQLite WAL is unavailable".to_owned()));
}
connection
.execute_batch("PRAGMA synchronous=FULL;")
.map_err(sql_issue)?;
let synchronous: i64 = connection
.query_row("PRAGMA synchronous", [], |row| row.get(0))
.map_err(sql_issue)?;
if synchronous != 2 {
return Err(OpenError::Fatal(
"SQLite FULL synchronization is unavailable".to_owned(),
));
}
Ok(connection)
}
fn validate_schema(connection: &Connection) -> Result<(), OpenError> {
let integrity: String = connection
.query_row("PRAGMA integrity_check", [], |row| row.get(0))
.map_err(sql_issue)?;
if integrity != "ok" {
return Err(OpenError::Rebuildable);
}
let mut statement = connection
.prepare("SELECT name, sql FROM sqlite_schema WHERE name NOT LIKE 'sqlite_%'")
.map_err(sql_issue)?;
let mut rows = statement.query([]).map_err(sql_issue)?;
let mut found = 0;
while let Some(row) = rows.next().map_err(sql_issue)? {
let name: String = row.get(0).map_err(|_| OpenError::Rebuildable)?;
let sql: String = row.get(1).map_err(|_| OpenError::Rebuildable)?;
match (name.as_str(), sql.as_str()) {
("metadata", METADATA_SQL)
| ("profiles", PROFILES_SQL)
| ("profiles_owner", OWNER_INDEX_SQL) => found += 1,
_ => return Err(OpenError::Rebuildable),
}
}
(found == 3).then_some(()).ok_or(OpenError::Rebuildable)
}
fn validate_rows(connection: &Connection) -> Result<Snapshot, OpenError> {
let count: i64 = connection
.query_row("SELECT count(*) FROM metadata", [], |row| row.get(0))
.map_err(sql_issue)?;
if count != 1 {
return Err(OpenError::Rebuildable);
}
let (version, cursor): (i64, Option<Vec<u8>>) = connection
.query_row(
"SELECT schema_version, last_applied_txid FROM metadata",
[],
|row| Ok((row.get(0)?, row.get(1)?)),
)
.map_err(|_| OpenError::Rebuildable)?;
if version != 1 {
return Err(OpenError::Rebuildable);
}
let cursor = cursor
.map(|bytes| bytes_12(bytes).map(TxId::from_bytes))
.transpose()
.map_err(|_| OpenError::Rebuildable)?;
let mut statement = connection
.prepare("SELECT profile_id, owner, revision, profile FROM profiles")
.map_err(sql_issue)?;
let mut rows = statement.query([]).map_err(sql_issue)?;
let mut profiles = Vec::new();
while let Some(row) = rows.next().map_err(sql_issue)? {
let raw = raw_profile(row).map_err(|_| OpenError::Rebuildable)?;
profiles.push(decode_saved(raw).map_err(|_| OpenError::Rebuildable)?);
}
if !profiles.is_empty() && cursor.is_none() {
return Err(OpenError::Rebuildable);
}
Ok(Snapshot { cursor, profiles })
}
fn sql_issue(error: rusqlite::Error) -> OpenError {
match error.sqlite_error_code() {
Some(ErrorCode::DatabaseCorrupt | ErrorCode::NotADatabase) => OpenError::Rebuildable,
_ => OpenError::Fatal(format!("SQLite is unavailable: {error}")),
}
}
fn remove_database(root: &Path) -> Result<(), String> {
for file in [
"profiles.sqlite3-wal",
"profiles.sqlite3-shm",
DATABASE_FILE,
] {
match fs::remove_file(root.join(file)) {
Ok(()) => (),
Err(error) if error.kind() == std::io::ErrorKind::NotFound => (),
Err(error) => return Err(format!("profile database cannot be rebuilt: {error}")),
}
}
Ok(())
}
#[cfg(test)]
mod tests {
use super::*;
use std::time::{SystemTime, UNIX_EPOCH};
#[test]
fn rejected_delete_cursor_survives_reopen_and_clear_resets_it() {
let nonce = SystemTime::now()
.duration_since(UNIX_EPOCH)
.unwrap()
.as_nanos();
let root = std::env::temp_dir().join(format!("profile-sqlite-test-{nonce}"));
let tx = |byte| TxId::from_bytes([byte; 12]);
let (database, _) = ProfileDatabase::open(&root).unwrap();
let action = ProfileAction::Delete {
profile_id: ProfileId::new(tx(1)),
actor: UserId::from_tx_id(tx(2)),
};
let outcome = database.apply(tx(3), action).unwrap();
let rejected = ApplyOutcome::Rejected("profile is unavailable".to_owned());
assert_eq!(outcome, rejected);
drop(database);
let (database, snapshot) = ProfileDatabase::open(&root).unwrap();
assert_eq!(snapshot.cursor(), Some(tx(3)));
assert!(snapshot.profiles().is_empty());
database.clear().unwrap();
drop(database);
let (database, snapshot) = ProfileDatabase::open(&root).unwrap();
assert_eq!(snapshot.cursor(), None);
assert!(snapshot.profiles().is_empty());
drop(database);
fs::remove_dir_all(root).unwrap();
}
}