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, 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 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::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::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::*;
fn txid(byte: u8) -> TxId {
TxId::from_bytes([byte; 12])
}
fn delete_action(actor: u8) -> ProfileAction {
ProfileAction::Delete {
profile_id: ProfileId::new(txid(1)),
actor: UserId::from_tx_id(txid(actor)),
}
}
#[test]
fn synchronous_callback_precedes_submission_error() {
let action = delete_action(2);
let callback = CallbackRecord {
txid: txid(3),
action: action.clone(),
outcome: ApplyOutcome::Rejected("callback rejection".to_owned()),
};
assert_eq!(
reconciliation_decision(
&action,
Some(&callback),
&Err("submission error".to_owned()),
),
ReconciliationDecision::Outcome(ApplyOutcome::Rejected(
"callback rejection".to_owned()
))
);
}
#[test]
fn callback_transaction_mismatch_faults() {
let action = delete_action(2);
let callback = CallbackRecord {
txid: txid(3),
action: action.clone(),
outcome: ApplyOutcome::Rejected("unused".to_owned()),
};
assert_eq!(
reconciliation_decision(&action, Some(&callback), &Ok(txid(4))),
ReconciliationDecision::Fault("profile callback transaction mismatch".to_owned())
);
}
#[test]
fn reorganization_invalidates_and_clears_correlations() {
let mut state = FacadeState::default();
state.pending.insert(
[7; 16],
PendingOperation {
expected: delete_action(2),
callback: None,
},
);
invalidate_state_for_reorg(&mut state);
assert_eq!(
state.fault.as_deref(),
Some("access profile facade invalidated by reorganization")
);
assert!(state.pending.is_empty());
}
}