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};
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_types::{decode_profile, encode_profile, resolve_built_in};
use kcode_k1_peering::K1Peering;
use kcode_k1_transaction::SubsystemId;
use kcode_k1_txn_ordering::{K1TxnOrdering, Subsystem};
const SUBSYSTEM_NAME: &str = "k1-profile-subsystem";
const WIRE_VERSION: u8 = 1;
const CREATE_TAG: u8 = 1;
const REPLACE_TAG: u8 = 2;
const DELETE_TAG: u8 = 3;
const HEADER_BYTES: usize = 2;
const OPERATION_ID_BYTES: usize = 16;
const TX_ID_BYTES: usize = 12;
const CREATE_PREFIX_BYTES: usize = HEADER_BYTES + OPERATION_ID_BYTES + TX_ID_BYTES;
const REPLACE_PREFIX_BYTES: usize = CREATE_PREFIX_BYTES + TX_ID_BYTES;
type OperationId = [u8; OPERATION_ID_BYTES];
#[derive(Debug, Eq, PartialEq)]
struct ParsedOperation {
operation_id: OperationId,
action: ProfileAction,
}
#[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()),
});
let handler = Arc::new(ProfileSubsystem {
inner: Arc::downgrade(&inner),
});
ordering.register_subsystem(subsystem, cursor, handler)?;
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 result = self.inner.store.get_for_user(user, profile_id);
let value = match result {
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 result = self.inner.store.list_for_user(user);
let value = match result {
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; OPERATION_ID_BYTES];
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 payload = match encode_operation(operation_id, &action) {
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()?;
if state.fault.is_none() {
state.fault =
Some("access profile facade invalidated by reorganization".to_owned());
}
state.pending.clear();
}
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 parsed = match parse_operation(payload) {
Ok(parsed) => parsed,
Err(error) => return Err(inner.fault(error)),
};
let action = parsed.action.clone();
let outcome = match inner.store.apply(id, parsed.action) {
Ok(outcome) => outcome,
Err(error) => return Err(inner.fault(error)),
};
inner.record_callback(parsed.operation_id, id, 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()
}
}
#[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()),
}
}
fn encode_operation(operation_id: OperationId, action: &ProfileAction) -> Result<Vec<u8>, String> {
match action {
ProfileAction::Create { owner, profile } => {
let profile_bytes = encode_profile(profile)?;
let mut payload = Vec::with_capacity(
CREATE_PREFIX_BYTES
.checked_add(profile_bytes.len())
.ok_or_else(|| "profile payload length overflow".to_owned())?,
);
payload.extend_from_slice(&[WIRE_VERSION, CREATE_TAG]);
payload.extend_from_slice(&operation_id);
payload.extend_from_slice(owner.as_tx_id().as_bytes());
payload.extend_from_slice(&profile_bytes);
Ok(payload)
}
ProfileAction::Replace {
profile_id,
actor,
profile,
} => {
let profile_bytes = encode_profile(profile)?;
let mut payload = Vec::with_capacity(
REPLACE_PREFIX_BYTES
.checked_add(profile_bytes.len())
.ok_or_else(|| "profile payload length overflow".to_owned())?,
);
payload.extend_from_slice(&[WIRE_VERSION, REPLACE_TAG]);
payload.extend_from_slice(&operation_id);
payload.extend_from_slice(profile_id.txid().as_bytes());
payload.extend_from_slice(actor.as_tx_id().as_bytes());
payload.extend_from_slice(&profile_bytes);
Ok(payload)
}
ProfileAction::Delete { profile_id, actor } => {
let mut payload = Vec::with_capacity(REPLACE_PREFIX_BYTES);
payload.extend_from_slice(&[WIRE_VERSION, DELETE_TAG]);
payload.extend_from_slice(&operation_id);
payload.extend_from_slice(profile_id.txid().as_bytes());
payload.extend_from_slice(actor.as_tx_id().as_bytes());
Ok(payload)
}
}
}
fn parse_operation(payload: &[u8]) -> Result<ParsedOperation, String> {
if payload.len() < HEADER_BYTES {
return Err("profile payload header is truncated".to_owned());
}
if payload[0] != WIRE_VERSION {
return Err("unsupported profile payload version".to_owned());
}
match payload[1] {
CREATE_TAG => parse_create(payload),
REPLACE_TAG => parse_replace(payload),
DELETE_TAG => parse_delete(payload),
_ => Err("unsupported profile payload action".to_owned()),
}
}
fn parse_create(payload: &[u8]) -> Result<ParsedOperation, String> {
if payload.len() <= CREATE_PREFIX_BYTES {
return Err("create profile payload is truncated".to_owned());
}
let operation_id = read_operation_id(payload);
let owner = UserId::from_tx_id(read_txid(&payload[18..30]));
let profile = decode_profile(&payload[CREATE_PREFIX_BYTES..])?;
Ok(ParsedOperation {
operation_id,
action: ProfileAction::Create { owner, profile },
})
}
fn parse_replace(payload: &[u8]) -> Result<ParsedOperation, String> {
if payload.len() <= REPLACE_PREFIX_BYTES {
return Err("replace profile payload is truncated".to_owned());
}
let operation_id = read_operation_id(payload);
let profile_id = ProfileId::new(read_txid(&payload[18..30]));
let actor = UserId::from_tx_id(read_txid(&payload[30..42]));
let profile = decode_profile(&payload[REPLACE_PREFIX_BYTES..])?;
Ok(ParsedOperation {
operation_id,
action: ProfileAction::Replace {
profile_id,
actor,
profile,
},
})
}
fn parse_delete(payload: &[u8]) -> Result<ParsedOperation, String> {
if payload.len() != REPLACE_PREFIX_BYTES {
return Err("delete profile payload must be exactly 42 bytes".to_owned());
}
Ok(ParsedOperation {
operation_id: read_operation_id(payload),
action: ProfileAction::Delete {
profile_id: ProfileId::new(read_txid(&payload[18..30])),
actor: UserId::from_tx_id(read_txid(&payload[30..42])),
},
})
}
fn read_operation_id(payload: &[u8]) -> OperationId {
let mut operation_id = [0_u8; OPERATION_ID_BYTES];
operation_id.copy_from_slice(&payload[2..18]);
operation_id
}
fn read_txid(bytes: &[u8]) -> TxId {
let mut txid = [0_u8; TX_ID_BYTES];
txid.copy_from_slice(bytes);
TxId::from_bytes(txid)
}