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, ToSql, TransactionBehavior, params,
};
pub use kcode_k1_access_profile_records::{
ApplyOutcome, AuthorizationProfile, DEFAULT_PROFILE_NAME, ProfileAction, ProfileColor,
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=2), 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, color TEXT, 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=2), 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, color TEXT, 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(2, NULL);
COMMIT;";
const SELECT_PROFILE: &str =
"SELECT profile_id, owner, revision, name, color, profile FROM profiles";
#[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::CreateNamedColored { profile, .. }
| ProfileAction::Replace { profile, .. } => Some(encode_profile(profile)?),
ProfileAction::Delete { .. }
| ProfileAction::Rename { .. }
| ProfileAction::SetColor { .. } => 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,
None,
encoded
.as_deref()
.ok_or("profile encoding is unavailable")?,
)?,
ProfileAction::CreateNamed { owner, name, .. } => insert_profile(
tx,
callback,
owner,
name.as_str(),
None,
encoded
.as_deref()
.ok_or("profile encoding is unavailable")?,
)?,
ProfileAction::CreateNamedColored {
owner, name, color, ..
} => insert_profile(
tx,
callback,
owner,
name.as_str(),
color.as_ref().map(ProfileColor::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")?;
update_profile(tx, callback, profile_id, "profile", &blob)?
}
}
_ => unavailable(),
},
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))
}
_ => unavailable(),
},
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 {
update_profile(tx, callback, profile_id, "name", &name.as_str())?
}
}
_ => unavailable(),
},
ProfileAction::SetColor {
profile_id,
actor,
color,
} => match load_saved(tx, profile_id)? {
Some(current) if current.owner() == actor => {
if current.color() == color.as_ref() {
ApplyOutcome::Unchanged(current.revision())
} else {
let stored = color.as_ref().map(ProfileColor::as_str);
update_profile(tx, callback, profile_id, "color", &stored)?
}
}
_ => unavailable(),
},
};
let cursor = callback.into_bytes();
changed_one(tx.execute(
"UPDATE metadata SET last_applied_txid=?1",
params![cursor.as_slice()],
))?;
Ok(outcome)
}
fn update_profile(
tx: &rusqlite::Transaction<'_>,
callback: TxId,
profile_id: ProfileId,
column: &str,
value: &dyn ToSql,
) -> Result<ApplyOutcome, String> {
let sql = match column {
"profile" => "UPDATE profiles SET revision=?1, profile=?2 WHERE profile_id=?3",
"name" => "UPDATE profiles SET revision=?1, name=?2 WHERE profile_id=?3",
"color" => "UPDATE profiles SET revision=?1, color=?2 WHERE profile_id=?3",
_ => return Err("profile persistence contradiction".to_owned()),
};
let id = profile_id.txid().into_bytes();
let revision = callback.into_bytes();
changed_one(tx.execute(sql, params![revision.as_slice(), value, id.as_slice()]))?;
Ok(ApplyOutcome::Applied(ProfileRevision::new(
profile_id, callback,
)))
}
fn unavailable() -> ApplyOutcome {
ApplyOutcome::Rejected("profile is unavailable".to_owned())
}
fn insert_profile(
tx: &rusqlite::Transaction<'_>,
callback: TxId,
owner: UserId,
name: &str,
color: Option<&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, color, profile) VALUES(?1, ?2, ?3, ?4, ?5, ?6)",
params![id_bytes.as_slice(), owner_bytes.as_slice(), id_bytes.as_slice(), name, color, 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 select_one<P: rusqlite::Params>(
connection: &Connection,
sql: &str,
parameters: P,
) -> rusqlite::Result<Option<RawProfile>> {
connection
.query_row(sql, parameters, raw_profile)
.optional()
}
fn load_saved(
tx: &rusqlite::Transaction<'_>,
id: ProfileId,
) -> Result<Option<SavedProfile>, String> {
let bytes = id.txid().into_bytes();
select_one(
tx,
&format!("{SELECT_PROFILE} WHERE profile_id=?1"),
params![bytes.as_slice()],
)
.map_err(persistence_error)?
.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();
select_one(
&connection,
&format!("{SELECT_PROFILE} WHERE profile_id=?1 AND owner=?2"),
params![profile.as_slice(), owner.as_slice()],
)
.map_err(query_error)?
.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(&format!("{SELECT_PROFILE} 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,
color: Option<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)?,
color: row.get(4)?,
profile: row.get(5)?,
})
}
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 color = raw
.color
.map(ProfileColor::new)
.transpose()
.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_colored(
profile_id,
owner,
ProfileRevision::new(profile_id, revision),
name,
color,
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 empty: bool = connection
.query_row(
"SELECT NOT EXISTS(SELECT 1 FROM sqlite_schema WHERE name NOT LIKE 'sqlite_%')",
[],
|row| row.get(0),
)
.map_err(sql_issue)?;
if empty {
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 schema: (i64, i64) = connection
.query_row(
"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_%'",
params![METADATA_SQL, PROFILES_SQL, OWNER_INDEX_SQL],
|row| Ok((row.get(0)?, row.get(1)?)),
)
.map_err(sql_issue)?;
if schema != (3, 3) {
return Err(OpenError::Rebuildable);
}
Ok(())
}
fn validate_rows(connection: &Connection) -> Result<Snapshot, OpenError> {
let (count, version, cursor): (i64, i64, Option<Vec<u8>>) = connection
.query_row(
"SELECT (SELECT count(*) FROM metadata), schema_version, last_applied_txid FROM metadata",
[],
|row| Ok((row.get(0)?, row.get(1)?, row.get(2)?)),
)
.map_err(|_| OpenError::Rebuildable)?;
if count != 1 || version != 2 {
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).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,
] {
if let Err(error) = fs::remove_file(root.join(file))
&& error.kind() != std::io::ErrorKind::NotFound
{
return Err(format!("profile database cannot be rebuilt: {error}"));
}
}
Ok(())
}
#[cfg(test)]
mod tests;