use std::{
fs,
path::{Path, PathBuf},
sync::Mutex,
time::Duration,
};
use kcode_k1_access_profile_codec::{decode_profile, encode_profile};
use rusqlite::{
Connection, ErrorCode, OpenFlags, OptionalExtension, Row, TransactionBehavior, params,
};
pub use kcode_k1_access_profile_records::{
ApplyOutcome, AuthorizationProfile, DEFAULT_PROFILE_NAME, ProfileAction, ProfileId,
ProfileName, ProfileRevision, SavedProfile, 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), name TEXT NOT NULL, 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), name TEXT NOT NULL, 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 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::CreateNamed { profile, .. }
| ProfileAction::Replace { profile, .. } => Some(encode_profile(profile)?),
ProfileAction::Delete { .. } | ProfileAction::Rename { .. } => 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, .. } => insert_profile(
tx,
callback,
owner,
DEFAULT_PROFILE_NAME,
encoded
.as_deref()
.ok_or("profile encoding is unavailable")?,
)?,
ProfileAction::CreateNamed { owner, name, .. } => insert_profile(
tx,
callback,
owner,
name.as_str(),
encoded
.as_deref()
.ok_or("profile encoding is unavailable")?,
)?,
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()),
},
ProfileAction::Rename {
profile_id,
actor,
name,
} => match load_saved(tx, profile_id)? {
Some(current) if current.owner() == actor => {
if current.name() == &name {
ApplyOutcome::Unchanged(current.revision())
} else {
let id = profile_id.txid().into_bytes();
let revision = callback.into_bytes();
changed_one(tx.execute(
"UPDATE profiles SET revision=?1, name=?2 WHERE profile_id=?3",
params![revision.as_slice(), name.as_str(), 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 insert_profile(
tx: &rusqlite::Transaction<'_>,
callback: TxId,
owner: UserId,
name: &str,
profile: &[u8],
) -> Result<ApplyOutcome, String> {
let id = ProfileId::new(callback);
if load_saved(tx, id)?.is_some() {
return Ok(ApplyOutcome::Rejected("profile already exists".to_owned()));
}
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, name, profile) VALUES(?1, ?2, ?3, ?4, ?5)",
params![
id_bytes.as_slice(),
owner_bytes.as_slice(),
id_bytes.as_slice(),
name,
profile
],
))?;
Ok(ApplyOutcome::Applied(ProfileRevision::new(id, callback)))
}
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, name, 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, name, 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, name, 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>,
name: String,
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)?,
name: row.get(3)?,
profile: row.get(4)?,
})
}
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 name = ProfileName::new(raw.name).map_err(|_| "profile persistence contradiction")?;
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::new_named(
profile_id,
owner,
ProfileRevision::new(profile_id, revision),
name,
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, name, 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 kcode_k1_access_profile_values::ProfileOwner;
fn tx(byte: u8) -> TxId {
TxId::from_bytes([byte; 12])
}
fn profile() -> AuthorizationProfile {
AuthorizationProfile::new(vec![ProfileOwner::RequestUser], Vec::new()).unwrap()
}
#[test]
fn persistence_smoke() {
let root = std::env::temp_dir().join(format!("profile-sqlite-test-{}", std::process::id()));
fs::remove_dir_all(&root).ok();
let owner = UserId::from_tx_id(tx(1));
let named = ProfileName::new("Production".to_owned()).unwrap();
let next = ProfileName::new("Renamed".to_owned()).unwrap();
let id = ProfileId::new(tx(2));
let (database, _) = ProfileDatabase::open(&root).unwrap();
let action = ProfileAction::CreateNamed {
owner,
name: named.clone(),
profile: profile(),
};
database.apply(tx(2), action).unwrap();
let saved = database.list_for_user(owner).unwrap().pop().unwrap();
assert_eq!(saved.name(), &named);
let action = ProfileAction::Rename {
profile_id: id,
actor: owner,
name: next.clone(),
};
database.apply(tx(3), action).unwrap();
let saved = database.list_for_user(owner).unwrap().pop().unwrap();
assert_eq!(
(saved.name(), saved.revision()),
(&next, ProfileRevision::new(id, tx(3)))
);
let action = ProfileAction::Replace {
profile_id: id,
actor: owner,
profile: profile(),
};
database.apply(tx(4), action).unwrap();
assert_eq!(database.list_for_user(owner).unwrap(), vec![saved.clone()]);
let (database, snapshot) = ProfileDatabase::open(&root).unwrap();
assert_eq!(
(snapshot.cursor(), snapshot.profiles()),
(Some(tx(4)), std::slice::from_ref(&saved))
);
database.clear().unwrap();
let (_, snapshot) = ProfileDatabase::open(&root).unwrap();
assert_eq!(
snapshot,
Snapshot {
cursor: None,
profiles: Vec::new()
}
);
fs::remove_dir_all(root).unwrap();
}
}