1use std::collections::HashMap;
2use std::path::Path;
3use std::sync::{Arc, Mutex, MutexGuard, Weak};
4
5pub use kcode_k1_access_profile_store::SavedProfile;
6use kcode_k1_access_profile_store::{ApplyOutcome, ProfileAction, ProfileStore};
7use kcode_k1_access_profile_types::resolve_built_in;
8pub use kcode_k1_access_profile_types::{
9 AuthorizationProfile, Authorizations, GroupId, ModelId, OwnerSubject, ProfileId, ProfileName,
10 ProfileOwner, ProfileRevision, ProfileSelection, ProfileSource, ProfileViewer,
11 RequestPrincipal, ResolvedProfile, TxId, UserId, ViewerSubject,
12};
13use kcode_k1_access_profile_wire::{
14 ProfileMutation, ProfileOperation, encode_operation, parse_operation,
15};
16use kcode_k1_peering::K1Peering;
17use kcode_k1_transaction::SubsystemId;
18use kcode_k1_txn_ordering::{K1TxnOrdering, Subsystem};
19
20const SUBSYSTEM_NAME: &str = "k1-profile-subsystem";
21type OperationId = [u8; 16];
22
23#[derive(Clone, Debug, Eq, PartialEq)]
24struct CallbackRecord {
25 txid: TxId,
26 action: ProfileAction,
27 outcome: ApplyOutcome,
28}
29
30struct PendingOperation {
31 expected: ProfileAction,
32 callback: Option<CallbackRecord>,
33}
34
35#[derive(Default)]
36struct FacadeState {
37 fault: Option<String>,
38 pending: HashMap<OperationId, PendingOperation>,
39}
40
41struct Inner {
42 store: ProfileStore,
43 peering: Arc<K1Peering>,
44 subsystem: SubsystemId,
45 state: Mutex<FacadeState>,
46}
47
48struct ProfileSubsystem {
49 inner: Weak<Inner>,
50}
51
52pub struct K1AccessProfiles {
53 inner: Arc<Inner>,
54}
55
56impl K1AccessProfiles {
57 pub fn open(
58 root: &Path,
59 ordering: Arc<K1TxnOrdering>,
60 peering: Arc<K1Peering>,
61 ) -> Result<Self, String> {
62 let (store, cursor) = ProfileStore::open(root, ordering.clone())?;
63 let subsystem = SubsystemId::from_str(SUBSYSTEM_NAME)?;
64 let inner = Arc::new(Inner {
65 store,
66 peering,
67 subsystem,
68 state: Mutex::new(FacadeState::default()),
69 });
70 ordering.register_subsystem(
71 subsystem,
72 cursor,
73 Arc::new(ProfileSubsystem {
74 inner: Arc::downgrade(&inner),
75 }),
76 )?;
77 Ok(Self { inner })
78 }
79
80 pub fn create(
81 &self,
82 owner: UserId,
83 profile: AuthorizationProfile,
84 ) -> Result<ProfileRevision, String> {
85 self.submit(ProfileAction::Create { owner, profile })
86 }
87
88 pub fn create_named(
89 &self,
90 owner: UserId,
91 name: ProfileName,
92 profile: AuthorizationProfile,
93 ) -> Result<ProfileRevision, String> {
94 self.submit(ProfileAction::CreateNamed {
95 owner,
96 name,
97 profile,
98 })
99 }
100
101 pub fn rename(
102 &self,
103 actor: UserId,
104 profile_id: ProfileId,
105 name: ProfileName,
106 ) -> Result<ProfileRevision, String> {
107 self.submit(ProfileAction::Rename {
108 profile_id,
109 actor,
110 name,
111 })
112 }
113
114 pub fn replace(
115 &self,
116 actor: UserId,
117 profile_id: ProfileId,
118 profile: AuthorizationProfile,
119 ) -> Result<ProfileRevision, String> {
120 self.submit(ProfileAction::Replace {
121 profile_id,
122 actor,
123 profile,
124 })
125 }
126
127 pub fn delete(&self, actor: UserId, profile_id: ProfileId) -> Result<ProfileRevision, String> {
128 self.submit(ProfileAction::Delete { profile_id, actor })
129 }
130
131 pub fn get_for_user(
132 &self,
133 user: UserId,
134 profile_id: ProfileId,
135 ) -> Result<Option<SavedProfile>, String> {
136 self.inner.ready()?;
137 let value = match self.inner.store.get_for_user(user, profile_id) {
138 Ok(value) => value,
139 Err(error) => return Err(self.inner.fault(error)),
140 };
141 self.inner.ready()?;
142 Ok(value)
143 }
144
145 pub fn list_for_user(&self, user: UserId) -> Result<Vec<SavedProfile>, String> {
146 self.inner.ready()?;
147 let value = match self.inner.store.list_for_user(user) {
148 Ok(value) => value,
149 Err(error) => return Err(self.inner.fault(error)),
150 };
151 self.inner.ready()?;
152 Ok(value)
153 }
154
155 pub fn resolve(
156 &self,
157 principal: RequestPrincipal,
158 selection: ProfileSelection,
159 ) -> Result<ResolvedProfile, String> {
160 match selection {
161 ProfileSelection::BuiltIn => resolve_built_in(principal),
162 ProfileSelection::Inline(profile) => {
163 let authorizations = profile.resolve(principal)?;
164 ResolvedProfile::new(authorizations, ProfileSource::Inline, None)
165 }
166 ProfileSelection::Saved(profile_id) => {
167 let saved = self
168 .get_for_user(principal.user(), profile_id)?
169 .ok_or_else(|| "profile is unavailable".to_owned())?;
170 let authorizations = saved.profile().resolve(principal)?;
171 ResolvedProfile::new(
172 authorizations,
173 ProfileSource::Saved(profile_id),
174 Some(saved.revision()),
175 )
176 }
177 }
178 }
179
180 fn submit(&self, action: ProfileAction) -> Result<ProfileRevision, String> {
181 self.inner.ready()?;
182 let operation_id = loop {
183 let mut operation_id = [0_u8; 16];
184 getrandom::fill(&mut operation_id).map_err(|error| error.to_string())?;
185 match self.inner.reserve(operation_id, action.clone()) {
186 Ok(()) => break operation_id,
187 Err(error) if error == "operation ID collision" => continue,
188 Err(error) => return Err(error),
189 }
190 };
191 let operation = ProfileOperation::new(operation_id, mutation_from_action(&action));
192 let payload = match encode_operation(&operation) {
193 Ok(payload) => payload,
194 Err(error) => {
195 self.inner.cancel(operation_id)?;
196 return Err(error);
197 }
198 };
199 let submission = self
200 .inner
201 .peering
202 .submit_txn(self.inner.subsystem, &payload);
203 self.inner.reconcile(operation_id, submission)
204 }
205}
206
207impl Inner {
208 fn lock_state(&self) -> Result<MutexGuard<'_, FacadeState>, String> {
209 self.state
210 .lock()
211 .map_err(|_| "access profile facade state lock is poisoned".to_owned())
212 }
213
214 fn ready(&self) -> Result<(), String> {
215 let state = self.lock_state()?;
216 match &state.fault {
217 Some(error) => Err(error.clone()),
218 None => Ok(()),
219 }
220 }
221
222 fn fault(&self, error: String) -> String {
223 let Ok(mut state) = self.state.lock() else {
224 return "access profile facade state lock is poisoned".to_owned();
225 };
226 if let Some(existing) = &state.fault {
227 return existing.clone();
228 }
229 state.fault = Some(error.clone());
230 error
231 }
232
233 fn reserve(&self, operation_id: OperationId, expected: ProfileAction) -> Result<(), String> {
234 let mut state = self.lock_state()?;
235 if let Some(error) = &state.fault {
236 return Err(error.clone());
237 }
238 if state.pending.contains_key(&operation_id) {
239 return Err("operation ID collision".to_owned());
240 }
241 state.pending.insert(
242 operation_id,
243 PendingOperation {
244 expected,
245 callback: None,
246 },
247 );
248 Ok(())
249 }
250
251 fn cancel(&self, operation_id: OperationId) -> Result<(), String> {
252 let mut state = self.lock_state()?;
253 if let Some(error) = &state.fault {
254 return Err(error.clone());
255 }
256 state.pending.remove(&operation_id);
257 Ok(())
258 }
259
260 fn record_callback(
261 &self,
262 operation_id: OperationId,
263 txid: TxId,
264 action: ProfileAction,
265 outcome: ApplyOutcome,
266 ) -> Result<(), String> {
267 let mut state = self.lock_state()?;
268 if let Some(error) = &state.fault {
269 return Err(error.clone());
270 }
271 let issue = match state.pending.get_mut(&operation_id) {
272 Some(pending) if pending.callback.is_some() => Some("duplicate profile callback"),
273 Some(pending) => {
274 let mismatch = pending.expected != action;
275 pending.callback = Some(CallbackRecord {
276 txid,
277 action,
278 outcome,
279 });
280 mismatch.then_some("profile callback action mismatch")
281 }
282 None => None,
283 };
284 if let Some(issue) = issue {
285 let error = issue.to_owned();
286 state.fault = Some(error.clone());
287 return Err(error);
288 }
289 Ok(())
290 }
291
292 fn reconcile(
293 &self,
294 operation_id: OperationId,
295 submission: Result<TxId, String>,
296 ) -> Result<ProfileRevision, String> {
297 let (pending, fault) = {
298 let mut state = self.lock_state()?;
299 (state.pending.remove(&operation_id), state.fault.clone())
300 };
301 let Some(pending) = pending else {
302 return Err(
303 fault.unwrap_or_else(|| "pending profile operation is unavailable".to_owned())
304 );
305 };
306 if pending.callback.is_none()
307 && let Some(error) = fault
308 {
309 return Err(error);
310 }
311 match reconciliation_decision(&pending.expected, pending.callback.as_ref(), &submission) {
312 ReconciliationDecision::Outcome(ApplyOutcome::Applied(revision))
313 | ReconciliationDecision::Outcome(ApplyOutcome::Unchanged(revision)) => Ok(revision),
314 ReconciliationDecision::Outcome(ApplyOutcome::Rejected(error))
315 | ReconciliationDecision::Error(error) => Err(error),
316 ReconciliationDecision::Fault(error) => Err(self.fault(error)),
317 }
318 }
319
320 fn invalidate_for_reorg(&self) -> Result<(), String> {
321 {
322 let mut state = self.lock_state()?;
323 invalidate_state_for_reorg(&mut state);
324 }
325 self.store.clear()
326 }
327}
328
329impl Subsystem for ProfileSubsystem {
330 fn submit_txn(&self, id: TxId, payload: &[u8]) -> Result<(), String> {
331 let inner = self
332 .inner
333 .upgrade()
334 .ok_or_else(|| "access profile facade is unavailable".to_owned())?;
335 inner.ready()?;
336 let operation = match parse_operation(payload) {
337 Ok(operation) => operation,
338 Err(error) => return Err(inner.fault(error)),
339 };
340 let operation_id = operation.operation_id();
341 let action = action_from_mutation(operation.mutation().clone());
342 let callback_action = action.clone();
343 let outcome = match inner.store.apply(id, action) {
344 Ok(outcome) => outcome,
345 Err(error) => return Err(inner.fault(error)),
346 };
347 inner.record_callback(operation_id, id, callback_action, outcome)
348 }
349
350 fn reorg(&self) -> Result<(), String> {
351 let inner = self
352 .inner
353 .upgrade()
354 .ok_or_else(|| "access profile facade is unavailable".to_owned())?;
355 inner.invalidate_for_reorg()
356 }
357}
358
359fn mutation_from_action(action: &ProfileAction) -> ProfileMutation {
360 match action {
361 ProfileAction::Create { owner, profile } => ProfileMutation::Create {
362 owner: *owner,
363 profile: profile.clone(),
364 },
365 ProfileAction::CreateNamed {
366 owner,
367 name,
368 profile,
369 } => ProfileMutation::CreateNamed {
370 owner: *owner,
371 name: name.clone(),
372 profile: profile.clone(),
373 },
374 ProfileAction::Rename {
375 profile_id,
376 actor,
377 name,
378 } => ProfileMutation::Rename {
379 profile_id: *profile_id,
380 actor: *actor,
381 name: name.clone(),
382 },
383 ProfileAction::Replace {
384 profile_id,
385 actor,
386 profile,
387 } => ProfileMutation::Replace {
388 profile_id: *profile_id,
389 actor: *actor,
390 profile: profile.clone(),
391 },
392 ProfileAction::Delete { profile_id, actor } => ProfileMutation::Delete {
393 profile_id: *profile_id,
394 actor: *actor,
395 },
396 }
397}
398
399fn action_from_mutation(mutation: ProfileMutation) -> ProfileAction {
400 match mutation {
401 ProfileMutation::Create { owner, profile } => ProfileAction::Create { owner, profile },
402 ProfileMutation::CreateNamed {
403 owner,
404 name,
405 profile,
406 } => ProfileAction::CreateNamed {
407 owner,
408 name,
409 profile,
410 },
411 ProfileMutation::Rename {
412 profile_id,
413 actor,
414 name,
415 } => ProfileAction::Rename {
416 profile_id,
417 actor,
418 name,
419 },
420 ProfileMutation::Replace {
421 profile_id,
422 actor,
423 profile,
424 } => ProfileAction::Replace {
425 profile_id,
426 actor,
427 profile,
428 },
429 ProfileMutation::Delete { profile_id, actor } => {
430 ProfileAction::Delete { profile_id, actor }
431 }
432 }
433}
434
435fn invalidate_state_for_reorg(state: &mut FacadeState) {
436 if state.fault.is_none() {
437 state.fault = Some("access profile facade invalidated by reorganization".to_owned());
438 }
439 state.pending.clear();
440}
441
442#[derive(Debug, Eq, PartialEq)]
443enum ReconciliationDecision {
444 Outcome(ApplyOutcome),
445 Error(String),
446 Fault(String),
447}
448
449fn reconciliation_decision(
450 expected: &ProfileAction,
451 callback: Option<&CallbackRecord>,
452 submission: &Result<TxId, String>,
453) -> ReconciliationDecision {
454 if let Some(callback) = callback {
455 if &callback.action != expected {
456 return ReconciliationDecision::Fault("profile callback action mismatch".to_owned());
457 }
458 if let Ok(submitted_txid) = submission
459 && *submitted_txid != callback.txid
460 {
461 return ReconciliationDecision::Fault(
462 "profile callback transaction mismatch".to_owned(),
463 );
464 }
465 return ReconciliationDecision::Outcome(callback.outcome.clone());
466 }
467 match submission {
468 Ok(_) => ReconciliationDecision::Fault(
469 "peering succeeded without matching profile callback".to_owned(),
470 ),
471 Err(error) => ReconciliationDecision::Error(error.clone()),
472 }
473}
474
475#[cfg(test)]
476mod tests {
477 use super::*;
478 use super::{PendingOperation as Pending, ProfileAction::*};
479
480 #[test]
481 fn mappings_and_reorganization() {
482 let user = UserId::from_tx_id(TxId::from_bytes([1; 12]));
483 let profile_id = ProfileId::new(TxId::from_bytes([2; 12]));
484 let name = ProfileName::new("Named profile".to_owned()).unwrap();
485 let profile =
486 AuthorizationProfile::new(vec![ProfileOwner::RequestUser], Vec::new()).unwrap();
487 let actions = [
488 Create {
489 owner: user,
490 profile: profile.clone(),
491 },
492 CreateNamed {
493 owner: user,
494 name: name.clone(),
495 profile: profile.clone(),
496 },
497 Replace {
498 profile_id,
499 actor: user,
500 profile,
501 },
502 Rename {
503 profile_id,
504 actor: user,
505 name,
506 },
507 Delete {
508 profile_id,
509 actor: user,
510 },
511 ];
512 let mut state = FacadeState::default();
513 for (i, action) in actions.into_iter().enumerate() {
514 assert_eq!(action_from_mutation(mutation_from_action(&action)), action);
515 state.pending.insert(
516 [i as u8; 16],
517 Pending {
518 expected: action,
519 callback: None,
520 },
521 );
522 }
523 invalidate_state_for_reorg(&mut state);
524 assert!(state.fault.is_some() && state.pending.is_empty());
525 }
526}