use std::{
fs,
io::ErrorKind,
path::{Path, PathBuf},
sync::{
Arc, Mutex,
atomic::{AtomicBool, Ordering},
},
time::{Duration, Instant},
};
use kcode_k1_access_profile_types::{decode_profile, encode_profile};
use kcode_k1_transaction::Transaction;
use kcode_k1_txn_ordering::K1TxnOrdering;
use rusqlite::{
Connection, ErrorCode, OpenFlags, OptionalExtension, Row, TransactionBehavior, params,
};
pub use kcode_k1_access_profile_types::{
AuthorizationProfile, ProfileId, ProfileRevision, TxId, UserId,
};
const DATABASE_FILE: &str = "profiles.sqlite3";
const SUBSYSTEM: &str = "k1-profile-subsystem";
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),
}
pub struct ProfileStore {
database: PathBuf,
ordering: Arc<K1TxnOrdering>,
apply_connection: Mutex<Connection>,
available: AtomicBool,
}
impl ProfileStore {
pub fn open(root: &Path, ordering: Arc<K1TxnOrdering>) -> Result<(Self, Option<TxId>), String> {
let started = Instant::now();
let result = Self::open_inner(root, ordering);
if started.elapsed() > Duration::from_millis(100) {
let outcome = if result.is_ok() { "ready" } else { "error" };
eprintln!(
"level=warn module=kcode-k1-access-profile-store operation=open elapsed_us={} outcome={outcome}",
started.elapsed().as_micros()
);
}
result
}
fn open_inner(
root: &Path,
ordering: Arc<K1TxnOrdering>,
) -> Result<(Self, Option<TxId>), String> {
fs::create_dir_all(root)
.map_err(|error| format!("profile root is unavailable: {error}"))?;
let database = root.join(DATABASE_FILE);
let loaded = load_database(&database, &ordering);
let (connection, cursor) = match loaded {
Ok(value) => value,
Err(OpenIssue::Fatal(error)) => return Err(error),
Err(OpenIssue::Corrupt) => {
remove_database(root)?;
match load_database(&database, &ordering) {
Ok(value) => value,
Err(OpenIssue::Fatal(error)) => return Err(error),
Err(OpenIssue::Corrupt) => {
return Err("profile database rebuild failed".to_owned());
}
}
}
};
Ok((
Self {
database,
ordering,
apply_connection: Mutex::new(connection),
available: AtomicBool::new(true),
},
cursor,
))
}
pub fn apply(
&self,
callback_txid: TxId,
action: ProfileAction,
) -> Result<ApplyOutcome, String> {
self.require_available()?;
match validate_transaction(&self.ordering, callback_txid) {
Ok(()) => {}
Err(OpenIssue::Corrupt) => {
return self.disable("profile callback contradicts KTO");
}
Err(OpenIssue::Fatal(error)) => return self.disable(error),
}
let encoded_profile = match &action {
ProfileAction::Create { profile, .. } | ProfileAction::Replace { profile, .. } => {
Some(encode_profile(profile)?)
}
ProfileAction::Delete { .. } => None,
};
let mut connection = match self.apply_connection.lock() {
Ok(connection) => connection,
Err(_) => return self.disable("profile apply lane is unavailable"),
};
self.require_available()?;
let transaction = match connection.transaction_with_behavior(TransactionBehavior::Immediate)
{
Ok(transaction) => transaction,
Err(error) => return self.disable(format!("profile persistence failed: {error}")),
};
let outcome = match apply_action(&transaction, callback_txid, action, encoded_profile) {
Ok(outcome) => outcome,
Err(error) => return self.disable(error),
};
if let Err(error) = transaction.commit() {
return self.disable(format!("profile persistence failed: {error}"));
}
Ok(outcome)
}
pub fn get_for_user(
&self,
user: UserId,
profile_id: ProfileId,
) -> Result<Option<SavedProfile>, String> {
self.require_available()?;
match query_profile(&self.database, user, profile_id) {
Ok(profile) => Ok(profile),
Err(error) => self.disable(error),
}
}
pub fn list_for_user(&self, user: UserId) -> Result<Vec<SavedProfile>, String> {
self.require_available()?;
match query_profiles(&self.database, user) {
Ok(profiles) => Ok(profiles),
Err(error) => self.disable(error),
}
}
pub fn clear(&self) -> Result<(), String> {
self.require_available()?;
let mut connection = match self.apply_connection.lock() {
Ok(connection) => connection,
Err(_) => return self.disable("profile apply lane is unavailable"),
};
self.require_available()?;
let transaction = match connection.transaction_with_behavior(TransactionBehavior::Immediate)
{
Ok(transaction) => transaction,
Err(error) => return self.disable(format!("profile persistence failed: {error}")),
};
if let Err(error) = transaction.execute("DELETE FROM profiles", []) {
return self.disable(format!("profile persistence failed: {error}"));
}
match transaction.execute("UPDATE metadata SET last_applied_txid=NULL", []) {
Ok(1) => {}
Ok(_) => return self.disable("profile persistence contradiction"),
Err(error) => return self.disable(format!("profile persistence failed: {error}")),
}
if let Err(error) = transaction.commit() {
return self.disable(format!("profile persistence failed: {error}"));
}
Ok(())
}
fn require_available(&self) -> Result<(), String> {
if self.available.load(Ordering::Acquire) {
Ok(())
} else {
Err("profile store is unavailable".to_owned())
}
}
fn disable<T>(&self, error: impl Into<String>) -> Result<T, String> {
self.available.store(false, Ordering::Release);
Err(error.into())
}
}
fn apply_action(
transaction: &rusqlite::Transaction<'_>,
callback_txid: TxId,
action: ProfileAction,
encoded_profile: Option<Vec<u8>>,
) -> Result<ApplyOutcome, String> {
let outcome = match action {
ProfileAction::Create { owner, .. } => {
let profile_id = ProfileId::new(callback_txid);
if load_saved(transaction, profile_id)?.is_some() {
ApplyOutcome::Rejected("profile already exists".to_owned())
} else {
let profile = encoded_profile
.as_deref()
.ok_or("profile encoding is unavailable")?;
let profile_id_bytes = callback_txid.into_bytes();
let owner_bytes = owner.as_tx_id().into_bytes();
let changed = transaction
.execute(
"INSERT INTO profiles(profile_id, owner, revision, profile) VALUES(?1, ?2, ?3, ?4)",
params![profile_id_bytes.as_slice(), owner_bytes.as_slice(), profile_id_bytes.as_slice(), profile],
)
.map_err(|error| format!("profile persistence failed: {error}"))?;
if changed != 1 {
return Err("profile persistence contradiction".to_owned());
}
ApplyOutcome::Applied(ProfileRevision::new(profile_id, callback_txid))
}
}
ProfileAction::Replace {
profile_id,
actor,
profile,
} => match load_saved(transaction, profile_id)? {
Some(current) if current.owner() == actor => {
if current.profile() == &profile {
ApplyOutcome::Unchanged(current.revision())
} else {
let profile = encoded_profile
.as_deref()
.ok_or("profile encoding is unavailable")?;
let profile_id_bytes = profile_id.txid().into_bytes();
let revision_bytes = callback_txid.into_bytes();
let changed = transaction
.execute(
"UPDATE profiles SET revision=?1, profile=?2 WHERE profile_id=?3",
params![
revision_bytes.as_slice(),
profile,
profile_id_bytes.as_slice()
],
)
.map_err(|error| format!("profile persistence failed: {error}"))?;
if changed != 1 {
return Err("profile persistence contradiction".to_owned());
}
ApplyOutcome::Applied(ProfileRevision::new(profile_id, callback_txid))
}
}
_ => ApplyOutcome::Rejected("profile is unavailable".to_owned()),
},
ProfileAction::Delete { profile_id, actor } => match load_saved(transaction, profile_id)? {
Some(current) if current.owner() == actor => {
let profile_id_bytes = profile_id.txid().into_bytes();
let changed = transaction
.execute(
"DELETE FROM profiles WHERE profile_id=?1",
params![profile_id_bytes.as_slice()],
)
.map_err(|error| format!("profile persistence failed: {error}"))?;
if changed != 1 {
return Err("profile persistence contradiction".to_owned());
}
ApplyOutcome::Applied(ProfileRevision::new(profile_id, callback_txid))
}
_ => ApplyOutcome::Rejected("profile is unavailable".to_owned()),
},
};
let cursor = callback_txid.into_bytes();
let changed = transaction
.execute(
"UPDATE metadata SET last_applied_txid=?1",
params![cursor.as_slice()],
)
.map_err(|error| format!("profile persistence failed: {error}"))?;
if changed != 1 {
return Err("profile persistence contradiction".to_owned());
}
Ok(outcome)
}
fn load_saved(
transaction: &rusqlite::Transaction<'_>,
profile_id: ProfileId,
) -> Result<Option<SavedProfile>, String> {
let profile_id_bytes = profile_id.txid().into_bytes();
let raw = transaction
.query_row(
"SELECT profile_id, owner, revision, profile FROM profiles WHERE profile_id=?1",
params![profile_id_bytes.as_slice()],
raw_profile,
)
.optional()
.map_err(|error| format!("profile persistence failed: {error}"))?;
raw.map(decode_saved).transpose()
}
fn query_profile(
database: &Path,
user: UserId,
profile_id: ProfileId,
) -> Result<Option<SavedProfile>, String> {
let connection = open_query(database)?;
let user_bytes = user.as_tx_id().into_bytes();
let profile_id_bytes = 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_id_bytes.as_slice(), user_bytes.as_slice()],
raw_profile,
)
.optional()
.map_err(|error| format!("profile query failed: {error}"))?;
raw.map(decode_saved).transpose()
}
fn query_profiles(database: &Path, user: UserId) -> Result<Vec<SavedProfile>, String> {
let connection = open_query(database)?;
let user_bytes = user.as_tx_id().into_bytes();
let mut statement = connection
.prepare("SELECT profile_id, owner, revision, profile FROM profiles WHERE owner=?1")
.map_err(|error| format!("profile query failed: {error}"))?;
let mut rows = statement
.query(params![user_bytes.as_slice()])
.map_err(|error| format!("profile query failed: {error}"))?;
let mut profiles = Vec::new();
while let Some(row) = rows
.next()
.map_err(|error| format!("profile query failed: {error}"))?
{
let raw = raw_profile(row).map_err(|error| format!("profile query failed: {error}"))?;
profiles.push(decode_saved(raw)?);
}
Ok(profiles)
}
fn open_query(database: &Path) -> Result<Connection, String> {
let connection = Connection::open_with_flags(
database,
OpenFlags::SQLITE_OPEN_READ_ONLY | OpenFlags::SQLITE_OPEN_NO_MUTEX,
)
.map_err(|error| format!("profile query failed: {error}"))?;
connection
.busy_timeout(Duration::ZERO)
.map_err(|error| format!("profile query failed: {error}"))?;
Ok(connection)
}
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 = TxId::from_bytes(bytes_12(raw.revision)?);
let profile = decode_profile(&raw.profile).map_err(|_| "profile persistence contradiction")?;
Ok(SavedProfile {
profile_id,
owner,
revision: ProfileRevision::new(profile_id, revision_txid),
profile,
})
}
fn bytes_12(bytes: Vec<u8>) -> Result<[u8; 12], String> {
bytes
.try_into()
.map_err(|_| "profile persistence contradiction".to_owned())
}
enum OpenIssue {
Corrupt,
Fatal(String),
}
fn load_database(
database: &Path,
ordering: &K1TxnOrdering,
) -> Result<(Connection, Option<TxId>), OpenIssue> {
let connection = connect(database)?;
let object_count: i64 = connection
.query_row(
"SELECT count(*) FROM sqlite_schema WHERE name NOT LIKE 'sqlite_%'",
[],
|row| row.get(0),
)
.map_err(sql_issue)?;
if object_count == 0 {
connection.execute_batch(CREATE_SCHEMA).map_err(sql_issue)?;
}
validate_schema(&connection)?;
let cursor = validate_rows(&connection, ordering)?;
Ok((connection, cursor))
}
fn connect(database: &Path) -> Result<Connection, OpenIssue> {
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(OpenIssue::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(OpenIssue::Fatal(
"SQLite FULL synchronization is unavailable".to_owned(),
));
}
Ok(connection)
}
fn validate_schema(connection: &Connection) -> Result<(), OpenIssue> {
let integrity: String = connection
.query_row("PRAGMA integrity_check", [], |row| row.get(0))
.map_err(sql_issue)?;
if integrity != "ok" {
return Err(OpenIssue::Corrupt);
}
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 metadata = false;
let mut profiles = false;
let mut owner_index = false;
let mut count = 0;
while let Some(row) = rows.next().map_err(sql_issue)? {
let name: String = row.get(0).map_err(|_| OpenIssue::Corrupt)?;
let sql: String = row.get(1).map_err(|_| OpenIssue::Corrupt)?;
count += 1;
match name.as_str() {
"metadata" if sql == METADATA_SQL => metadata = true,
"profiles" if sql == PROFILES_SQL => profiles = true,
"profiles_owner" if sql == OWNER_INDEX_SQL => owner_index = true,
_ => return Err(OpenIssue::Corrupt),
}
}
if count != 3 || !metadata || !profiles || !owner_index {
return Err(OpenIssue::Corrupt);
}
Ok(())
}
fn validate_rows(
connection: &Connection,
ordering: &K1TxnOrdering,
) -> Result<Option<TxId>, OpenIssue> {
let metadata_count: i64 = connection
.query_row("SELECT count(*) FROM metadata", [], |row| row.get(0))
.map_err(sql_issue)?;
if metadata_count != 1 {
return Err(OpenIssue::Corrupt);
}
let (version, cursor_bytes): (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(|_| OpenIssue::Corrupt)?;
if version != 1 {
return Err(OpenIssue::Corrupt);
}
let cursor = cursor_bytes
.map(|bytes| bytes_12(bytes).map(TxId::from_bytes))
.transpose()
.map_err(|_| OpenIssue::Corrupt)?;
if let Some(id) = cursor {
validate_transaction(ordering, id)?;
}
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 any_profile = false;
while let Some(row) = rows.next().map_err(sql_issue)? {
any_profile = true;
let raw = raw_profile(row).map_err(|_| OpenIssue::Corrupt)?;
let saved = decode_saved(raw).map_err(|_| OpenIssue::Corrupt)?;
validate_transaction(ordering, saved.profile_id().txid())?;
validate_transaction(ordering, saved.revision().txid())?;
}
if any_profile && cursor.is_none() {
return Err(OpenIssue::Corrupt);
}
Ok(cursor)
}
fn validate_transaction(ordering: &K1TxnOrdering, id: TxId) -> Result<(), OpenIssue> {
let bytes = ordering
.get_txn(id)
.map_err(|error| OpenIssue::Fatal(format!("KTO query failed: {error}")))?
.ok_or(OpenIssue::Corrupt)?;
if !id.verify(&bytes) {
return Err(OpenIssue::Corrupt);
}
let transaction = Transaction::parse(&bytes).map_err(|_| OpenIssue::Corrupt)?;
if transaction.subsystem().as_str() != SUBSYSTEM {
return Err(OpenIssue::Corrupt);
}
Ok(())
}
fn sql_issue(error: rusqlite::Error) -> OpenIssue {
match error.sqlite_error_code() {
Some(ErrorCode::DatabaseCorrupt | ErrorCode::NotADatabase) => OpenIssue::Corrupt,
_ => OpenIssue::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() == ErrorKind::NotFound => {}
Err(error) => return Err(format!("profile database cannot be rebuilt: {error}")),
}
}
Ok(())
}