use std::collections::HashMap;
use std::path::Path;
use std::sync::{Arc, Mutex, MutexGuard, Weak};
pub use kcode_k1_access_profile_store::SavedProfile;
use kcode_k1_access_profile_store::{ApplyOutcome, ProfileAction, ProfileStore};
use kcode_k1_access_profile_types::resolve_built_in;
pub use kcode_k1_access_profile_types::{
AuthorizationProfile, Authorizations, GroupId, ModelId, OwnerSubject, ProfileId, ProfileName,
ProfileOwner, ProfileRevision, ProfileSelection, ProfileSource, ProfileViewer,
RequestPrincipal, ResolvedProfile, TxId, UserId, ViewerSubject,
};
use kcode_k1_access_profile_wire::{
ProfileMutation, ProfileOperation, encode_operation, parse_operation,
};
use kcode_k1_peering::K1Peering;
use kcode_k1_transaction::SubsystemId;
use kcode_k1_txn_ordering::{K1TxnOrdering, Subsystem};
const SUBSYSTEM_NAME: &str = "k1-profile-subsystem";
type OperationId = [u8; 16];
#[derive(Clone, Debug, Eq, PartialEq)]
struct CallbackRecord {
txid: TxId,
action: ProfileAction,
outcome: ApplyOutcome,
}
struct PendingOperation {
expected: ProfileAction,
callback: Option<CallbackRecord>,
}
#[derive(Default)]
struct FacadeState {
fault: Option<String>,
pending: HashMap<OperationId, PendingOperation>,
}
struct Inner {
store: ProfileStore,
peering: Arc<K1Peering>,
subsystem: SubsystemId,
state: Mutex<FacadeState>,
}
struct ProfileSubsystem {
inner: Weak<Inner>,
}
pub struct K1AccessProfiles {
inner: Arc<Inner>,
}
impl K1AccessProfiles {
pub fn open(
root: &Path,
ordering: Arc<K1TxnOrdering>,
peering: Arc<K1Peering>,
) -> Result<Self, String> {
let (store, cursor) = ProfileStore::open(root, ordering.clone())?;
let subsystem = SubsystemId::from_str(SUBSYSTEM_NAME)?;
let inner = Arc::new(Inner {
store,
peering,
subsystem,
state: Mutex::new(FacadeState::default()),
});
ordering.register_subsystem(
subsystem,
cursor,
Arc::new(ProfileSubsystem {
inner: Arc::downgrade(&inner),
}),
)?;
Ok(Self { inner })
}
pub fn create(
&self,
owner: UserId,
profile: AuthorizationProfile,
) -> Result<ProfileRevision, String> {
self.submit(ProfileAction::Create { owner, profile })
}
pub fn create_named(
&self,
owner: UserId,
name: ProfileName,
profile: AuthorizationProfile,
) -> Result<ProfileRevision, String> {
self.submit(ProfileAction::CreateNamed {
owner,
name,
profile,
})
}
pub fn rename(
&self,
actor: UserId,
profile_id: ProfileId,
name: ProfileName,
) -> Result<ProfileRevision, String> {
self.submit(ProfileAction::Rename {
profile_id,
actor,
name,
})
}
pub fn replace(
&self,
actor: UserId,
profile_id: ProfileId,
profile: AuthorizationProfile,
) -> Result<ProfileRevision, String> {
self.submit(ProfileAction::Replace {
profile_id,
actor,
profile,
})
}
pub fn delete(&self, actor: UserId, profile_id: ProfileId) -> Result<ProfileRevision, String> {
self.submit(ProfileAction::Delete { profile_id, actor })
}
pub fn get_for_user(
&self,
user: UserId,
profile_id: ProfileId,
) -> Result<Option<SavedProfile>, String> {
self.inner.ready()?;
let value = match self.inner.store.get_for_user(user, profile_id) {
Ok(value) => value,
Err(error) => return Err(self.inner.fault(error)),
};
self.inner.ready()?;
Ok(value)
}
pub fn list_for_user(&self, user: UserId) -> Result<Vec<SavedProfile>, String> {
self.inner.ready()?;
let value = match self.inner.store.list_for_user(user) {
Ok(value) => value,
Err(error) => return Err(self.inner.fault(error)),
};
self.inner.ready()?;
Ok(value)
}
pub fn resolve(
&self,
principal: RequestPrincipal,
selection: ProfileSelection,
) -> Result<ResolvedProfile, String> {
match selection {
ProfileSelection::BuiltIn => resolve_built_in(principal),
ProfileSelection::Inline(profile) => {
let authorizations = profile.resolve(principal)?;
ResolvedProfile::new(authorizations, ProfileSource::Inline, None)
}
ProfileSelection::Saved(profile_id) => {
let saved = self
.get_for_user(principal.user(), profile_id)?
.ok_or_else(|| "profile is unavailable".to_owned())?;
let authorizations = saved.profile().resolve(principal)?;
ResolvedProfile::new(
authorizations,
ProfileSource::Saved(profile_id),
Some(saved.revision()),
)
}
}
}
fn submit(&self, action: ProfileAction) -> Result<ProfileRevision, String> {
self.inner.ready()?;
let operation_id = loop {
let mut operation_id = [0_u8; 16];
getrandom::fill(&mut operation_id).map_err(|error| error.to_string())?;
match self.inner.reserve(operation_id, action.clone()) {
Ok(()) => break operation_id,
Err(error) if error == "operation ID collision" => continue,
Err(error) => return Err(error),
}
};
let operation = ProfileOperation::new(operation_id, mutation_from_action(&action));
let payload = match encode_operation(&operation) {
Ok(payload) => payload,
Err(error) => {
self.inner.cancel(operation_id)?;
return Err(error);
}
};
let submission = self
.inner
.peering
.submit_txn(self.inner.subsystem, &payload);
self.inner.reconcile(operation_id, submission)
}
}
impl Inner {
fn lock_state(&self) -> Result<MutexGuard<'_, FacadeState>, String> {
self.state
.lock()
.map_err(|_| "access profile facade state lock is poisoned".to_owned())
}
fn ready(&self) -> Result<(), String> {
let state = self.lock_state()?;
match &state.fault {
Some(error) => Err(error.clone()),
None => Ok(()),
}
}
fn fault(&self, error: String) -> String {
let Ok(mut state) = self.state.lock() else {
return "access profile facade state lock is poisoned".to_owned();
};
if let Some(existing) = &state.fault {
return existing.clone();
}
state.fault = Some(error.clone());
error
}
fn reserve(&self, operation_id: OperationId, expected: ProfileAction) -> Result<(), String> {
let mut state = self.lock_state()?;
if let Some(error) = &state.fault {
return Err(error.clone());
}
if state.pending.contains_key(&operation_id) {
return Err("operation ID collision".to_owned());
}
state.pending.insert(
operation_id,
PendingOperation {
expected,
callback: None,
},
);
Ok(())
}
fn cancel(&self, operation_id: OperationId) -> Result<(), String> {
let mut state = self.lock_state()?;
if let Some(error) = &state.fault {
return Err(error.clone());
}
state.pending.remove(&operation_id);
Ok(())
}
fn record_callback(
&self,
operation_id: OperationId,
txid: TxId,
action: ProfileAction,
outcome: ApplyOutcome,
) -> Result<(), String> {
let mut state = self.lock_state()?;
if let Some(error) = &state.fault {
return Err(error.clone());
}
let issue = match state.pending.get_mut(&operation_id) {
Some(pending) if pending.callback.is_some() => Some("duplicate profile callback"),
Some(pending) => {
let mismatch = pending.expected != action;
pending.callback = Some(CallbackRecord {
txid,
action,
outcome,
});
mismatch.then_some("profile callback action mismatch")
}
None => None,
};
if let Some(issue) = issue {
let error = issue.to_owned();
state.fault = Some(error.clone());
return Err(error);
}
Ok(())
}
fn reconcile(
&self,
operation_id: OperationId,
submission: Result<TxId, String>,
) -> Result<ProfileRevision, String> {
let (pending, fault) = {
let mut state = self.lock_state()?;
(state.pending.remove(&operation_id), state.fault.clone())
};
let Some(pending) = pending else {
return Err(
fault.unwrap_or_else(|| "pending profile operation is unavailable".to_owned())
);
};
if pending.callback.is_none()
&& let Some(error) = fault
{
return Err(error);
}
match reconciliation_decision(&pending.expected, pending.callback.as_ref(), &submission) {
ReconciliationDecision::Outcome(ApplyOutcome::Applied(revision))
| ReconciliationDecision::Outcome(ApplyOutcome::Unchanged(revision)) => Ok(revision),
ReconciliationDecision::Outcome(ApplyOutcome::Rejected(error))
| ReconciliationDecision::Error(error) => Err(error),
ReconciliationDecision::Fault(error) => Err(self.fault(error)),
}
}
fn invalidate_for_reorg(&self) -> Result<(), String> {
{
let mut state = self.lock_state()?;
invalidate_state_for_reorg(&mut state);
}
self.store.clear()
}
}
impl Subsystem for ProfileSubsystem {
fn submit_txn(&self, id: TxId, payload: &[u8]) -> Result<(), String> {
let inner = self
.inner
.upgrade()
.ok_or_else(|| "access profile facade is unavailable".to_owned())?;
inner.ready()?;
let operation = match parse_operation(payload) {
Ok(operation) => operation,
Err(error) => return Err(inner.fault(error)),
};
let operation_id = operation.operation_id();
let action = action_from_mutation(operation.mutation().clone());
let callback_action = action.clone();
let outcome = match inner.store.apply(id, action) {
Ok(outcome) => outcome,
Err(error) => return Err(inner.fault(error)),
};
inner.record_callback(operation_id, id, callback_action, outcome)
}
fn reorg(&self) -> Result<(), String> {
let inner = self
.inner
.upgrade()
.ok_or_else(|| "access profile facade is unavailable".to_owned())?;
inner.invalidate_for_reorg()
}
}
fn mutation_from_action(action: &ProfileAction) -> ProfileMutation {
match action {
ProfileAction::Create { owner, profile } => ProfileMutation::Create {
owner: *owner,
profile: profile.clone(),
},
ProfileAction::CreateNamed {
owner,
name,
profile,
} => ProfileMutation::CreateNamed {
owner: *owner,
name: name.clone(),
profile: profile.clone(),
},
ProfileAction::Rename {
profile_id,
actor,
name,
} => ProfileMutation::Rename {
profile_id: *profile_id,
actor: *actor,
name: name.clone(),
},
ProfileAction::Replace {
profile_id,
actor,
profile,
} => ProfileMutation::Replace {
profile_id: *profile_id,
actor: *actor,
profile: profile.clone(),
},
ProfileAction::Delete { profile_id, actor } => ProfileMutation::Delete {
profile_id: *profile_id,
actor: *actor,
},
}
}
fn action_from_mutation(mutation: ProfileMutation) -> ProfileAction {
match mutation {
ProfileMutation::Create { owner, profile } => ProfileAction::Create { owner, profile },
ProfileMutation::CreateNamed {
owner,
name,
profile,
} => ProfileAction::CreateNamed {
owner,
name,
profile,
},
ProfileMutation::Rename {
profile_id,
actor,
name,
} => ProfileAction::Rename {
profile_id,
actor,
name,
},
ProfileMutation::Replace {
profile_id,
actor,
profile,
} => ProfileAction::Replace {
profile_id,
actor,
profile,
},
ProfileMutation::Delete { profile_id, actor } => {
ProfileAction::Delete { profile_id, actor }
}
}
}
fn invalidate_state_for_reorg(state: &mut FacadeState) {
if state.fault.is_none() {
state.fault = Some("access profile facade invalidated by reorganization".to_owned());
}
state.pending.clear();
}
#[derive(Debug, Eq, PartialEq)]
enum ReconciliationDecision {
Outcome(ApplyOutcome),
Error(String),
Fault(String),
}
fn reconciliation_decision(
expected: &ProfileAction,
callback: Option<&CallbackRecord>,
submission: &Result<TxId, String>,
) -> ReconciliationDecision {
if let Some(callback) = callback {
if &callback.action != expected {
return ReconciliationDecision::Fault("profile callback action mismatch".to_owned());
}
if let Ok(submitted_txid) = submission
&& *submitted_txid != callback.txid
{
return ReconciliationDecision::Fault(
"profile callback transaction mismatch".to_owned(),
);
}
return ReconciliationDecision::Outcome(callback.outcome.clone());
}
match submission {
Ok(_) => ReconciliationDecision::Fault(
"peering succeeded without matching profile callback".to_owned(),
),
Err(error) => ReconciliationDecision::Error(error.clone()),
}
}
#[cfg(test)]
mod tests {
use super::*;
use super::{PendingOperation as Pending, ProfileAction::*};
#[test]
fn mappings_and_reorganization() {
let user = UserId::from_tx_id(TxId::from_bytes([1; 12]));
let profile_id = ProfileId::new(TxId::from_bytes([2; 12]));
let name = ProfileName::new("Named profile".to_owned()).unwrap();
let profile =
AuthorizationProfile::new(vec![ProfileOwner::RequestUser], Vec::new()).unwrap();
let actions = [
Create {
owner: user,
profile: profile.clone(),
},
CreateNamed {
owner: user,
name: name.clone(),
profile: profile.clone(),
},
Replace {
profile_id,
actor: user,
profile,
},
Rename {
profile_id,
actor: user,
name,
},
Delete {
profile_id,
actor: user,
},
];
let mut state = FacadeState::default();
for (i, action) in actions.into_iter().enumerate() {
assert_eq!(action_from_mutation(mutation_from_action(&action)), action);
state.pending.insert(
[i as u8; 16],
Pending {
expected: action,
callback: None,
},
);
}
invalidate_state_for_reorg(&mut state);
assert!(state.fault.is_some() && state.pending.is_empty());
}
}