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, ProfileColor, ProfileId,
10 ProfileName, 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 create_named_colored(
102 &self,
103 owner: UserId,
104 name: ProfileName,
105 color: Option<ProfileColor>,
106 profile: AuthorizationProfile,
107 ) -> Result<ProfileRevision, String> {
108 self.submit(ProfileAction::CreateNamedColored {
109 owner,
110 name,
111 color,
112 profile,
113 })
114 }
115
116 pub fn rename(
117 &self,
118 actor: UserId,
119 profile_id: ProfileId,
120 name: ProfileName,
121 ) -> Result<ProfileRevision, String> {
122 self.submit(ProfileAction::Rename {
123 profile_id,
124 actor,
125 name,
126 })
127 }
128
129 pub fn set_color(
130 &self,
131 actor: UserId,
132 profile_id: ProfileId,
133 color: Option<ProfileColor>,
134 ) -> Result<ProfileRevision, String> {
135 self.submit(ProfileAction::SetColor {
136 profile_id,
137 actor,
138 color,
139 })
140 }
141
142 pub fn replace(
143 &self,
144 actor: UserId,
145 profile_id: ProfileId,
146 profile: AuthorizationProfile,
147 ) -> Result<ProfileRevision, String> {
148 self.submit(ProfileAction::Replace {
149 profile_id,
150 actor,
151 profile,
152 })
153 }
154
155 pub fn delete(&self, actor: UserId, profile_id: ProfileId) -> Result<ProfileRevision, String> {
156 self.submit(ProfileAction::Delete { profile_id, actor })
157 }
158
159 pub fn get_for_user(
160 &self,
161 user: UserId,
162 profile_id: ProfileId,
163 ) -> Result<Option<SavedProfile>, String> {
164 self.inner.ready()?;
165 let value = match self.inner.store.get_for_user(user, profile_id) {
166 Ok(value) => value,
167 Err(error) => return Err(self.inner.fault(error)),
168 };
169 self.inner.ready()?;
170 Ok(value)
171 }
172
173 pub fn list_for_user(&self, user: UserId) -> Result<Vec<SavedProfile>, String> {
174 self.inner.ready()?;
175 let value = match self.inner.store.list_for_user(user) {
176 Ok(value) => value,
177 Err(error) => return Err(self.inner.fault(error)),
178 };
179 self.inner.ready()?;
180 Ok(value)
181 }
182
183 pub fn resolve(
184 &self,
185 principal: RequestPrincipal,
186 selection: ProfileSelection,
187 ) -> Result<ResolvedProfile, String> {
188 match selection {
189 ProfileSelection::BuiltIn => resolve_built_in(principal),
190 ProfileSelection::Inline(profile) => {
191 let authorizations = profile.resolve(principal)?;
192 ResolvedProfile::new(authorizations, ProfileSource::Inline, None)
193 }
194 ProfileSelection::Saved(profile_id) => {
195 let saved = self
196 .get_for_user(principal.user(), profile_id)?
197 .ok_or_else(|| "profile is unavailable".to_owned())?;
198 let authorizations = saved.profile().resolve(principal)?;
199 ResolvedProfile::new(
200 authorizations,
201 ProfileSource::Saved(profile_id),
202 Some(saved.revision()),
203 )
204 }
205 }
206 }
207
208 fn submit(&self, action: ProfileAction) -> Result<ProfileRevision, String> {
209 self.inner.ready()?;
210 let operation_id = loop {
211 let mut operation_id = [0_u8; 16];
212 getrandom::fill(&mut operation_id).map_err(|error| error.to_string())?;
213 match self.inner.reserve(operation_id, action.clone()) {
214 Ok(()) => break operation_id,
215 Err(error) if error == "operation ID collision" => continue,
216 Err(error) => return Err(error),
217 }
218 };
219 let operation = ProfileOperation::new(operation_id, mutation_from_action(&action));
220 let payload = match encode_operation(&operation) {
221 Ok(payload) => payload,
222 Err(error) => {
223 self.inner.cancel(operation_id)?;
224 return Err(error);
225 }
226 };
227 let submission = self
228 .inner
229 .peering
230 .submit_txn(self.inner.subsystem, &payload);
231 self.inner.reconcile(operation_id, submission)
232 }
233}
234
235impl Inner {
236 fn lock_state(&self) -> Result<MutexGuard<'_, FacadeState>, String> {
237 self.state
238 .lock()
239 .map_err(|_| "access profile facade state lock is poisoned".to_owned())
240 }
241
242 fn ready(&self) -> Result<(), String> {
243 let state = self.lock_state()?;
244 match &state.fault {
245 Some(error) => Err(error.clone()),
246 None => Ok(()),
247 }
248 }
249
250 fn fault(&self, error: String) -> String {
251 let Ok(mut state) = self.state.lock() else {
252 return "access profile facade state lock is poisoned".to_owned();
253 };
254 if let Some(existing) = &state.fault {
255 return existing.clone();
256 }
257 state.fault = Some(error.clone());
258 error
259 }
260
261 fn reserve(&self, operation_id: OperationId, expected: ProfileAction) -> Result<(), String> {
262 let mut state = self.lock_state()?;
263 if let Some(error) = &state.fault {
264 return Err(error.clone());
265 }
266 if state.pending.contains_key(&operation_id) {
267 return Err("operation ID collision".to_owned());
268 }
269 state.pending.insert(
270 operation_id,
271 PendingOperation {
272 expected,
273 callback: None,
274 },
275 );
276 Ok(())
277 }
278
279 fn cancel(&self, operation_id: OperationId) -> Result<(), String> {
280 let mut state = self.lock_state()?;
281 if let Some(error) = &state.fault {
282 return Err(error.clone());
283 }
284 state.pending.remove(&operation_id);
285 Ok(())
286 }
287
288 fn record_callback(
289 &self,
290 operation_id: OperationId,
291 txid: TxId,
292 action: ProfileAction,
293 outcome: ApplyOutcome,
294 ) -> Result<(), String> {
295 let mut state = self.lock_state()?;
296 if let Some(error) = &state.fault {
297 return Err(error.clone());
298 }
299 let issue = match state.pending.get_mut(&operation_id) {
300 Some(pending) if pending.callback.is_some() => Some("duplicate profile callback"),
301 Some(pending) => {
302 let mismatch = pending.expected != action;
303 pending.callback = Some(CallbackRecord {
304 txid,
305 action,
306 outcome,
307 });
308 mismatch.then_some("profile callback action mismatch")
309 }
310 None => None,
311 };
312 if let Some(issue) = issue {
313 let error = issue.to_owned();
314 state.fault = Some(error.clone());
315 return Err(error);
316 }
317 Ok(())
318 }
319
320 fn reconcile(
321 &self,
322 operation_id: OperationId,
323 submission: Result<TxId, String>,
324 ) -> Result<ProfileRevision, String> {
325 let (pending, fault) = {
326 let mut state = self.lock_state()?;
327 (state.pending.remove(&operation_id), state.fault.clone())
328 };
329 let Some(pending) = pending else {
330 return Err(
331 fault.unwrap_or_else(|| "pending profile operation is unavailable".to_owned())
332 );
333 };
334 if pending.callback.is_none()
335 && let Some(error) = fault
336 {
337 return Err(error);
338 }
339 match reconciliation_decision(&pending.expected, pending.callback.as_ref(), &submission) {
340 ReconciliationDecision::Outcome(ApplyOutcome::Applied(revision))
341 | ReconciliationDecision::Outcome(ApplyOutcome::Unchanged(revision)) => Ok(revision),
342 ReconciliationDecision::Outcome(ApplyOutcome::Rejected(error))
343 | ReconciliationDecision::Error(error) => Err(error),
344 ReconciliationDecision::Fault(error) => Err(self.fault(error)),
345 }
346 }
347
348 fn invalidate_for_reorg(&self) -> Result<(), String> {
349 {
350 let mut state = self.lock_state()?;
351 invalidate_state_for_reorg(&mut state);
352 }
353 self.store.clear()
354 }
355}
356
357impl Subsystem for ProfileSubsystem {
358 fn submit_txn(&self, id: TxId, payload: &[u8]) -> Result<(), String> {
359 let inner = self
360 .inner
361 .upgrade()
362 .ok_or_else(|| "access profile facade is unavailable".to_owned())?;
363 inner.ready()?;
364 let operation = match parse_operation(payload) {
365 Ok(operation) => operation,
366 Err(error) => return Err(inner.fault(error)),
367 };
368 let operation_id = operation.operation_id();
369 let action = action_from_mutation(operation.mutation().clone());
370 let callback_action = action.clone();
371 let outcome = match inner.store.apply(id, action) {
372 Ok(outcome) => outcome,
373 Err(error) => return Err(inner.fault(error)),
374 };
375 inner.record_callback(operation_id, id, callback_action, outcome)
376 }
377
378 fn reorg(&self) -> Result<(), String> {
379 let inner = self
380 .inner
381 .upgrade()
382 .ok_or_else(|| "access profile facade is unavailable".to_owned())?;
383 inner.invalidate_for_reorg()
384 }
385}
386
387fn mutation_from_action(action: &ProfileAction) -> ProfileMutation {
388 match action {
389 ProfileAction::Create { owner, profile } => ProfileMutation::Create {
390 owner: *owner,
391 profile: profile.clone(),
392 },
393 ProfileAction::CreateNamed {
394 owner,
395 name,
396 profile,
397 } => ProfileMutation::CreateNamed {
398 owner: *owner,
399 name: name.clone(),
400 profile: profile.clone(),
401 },
402 ProfileAction::CreateNamedColored {
403 owner,
404 name,
405 color,
406 profile,
407 } => ProfileMutation::CreateNamedColored {
408 owner: *owner,
409 name: name.clone(),
410 color: color.clone(),
411 profile: profile.clone(),
412 },
413 ProfileAction::Rename {
414 profile_id,
415 actor,
416 name,
417 } => ProfileMutation::Rename {
418 profile_id: *profile_id,
419 actor: *actor,
420 name: name.clone(),
421 },
422 ProfileAction::SetColor {
423 profile_id,
424 actor,
425 color,
426 } => ProfileMutation::SetColor {
427 profile_id: *profile_id,
428 actor: *actor,
429 color: color.clone(),
430 },
431 ProfileAction::Replace {
432 profile_id,
433 actor,
434 profile,
435 } => ProfileMutation::Replace {
436 profile_id: *profile_id,
437 actor: *actor,
438 profile: profile.clone(),
439 },
440 ProfileAction::Delete { profile_id, actor } => ProfileMutation::Delete {
441 profile_id: *profile_id,
442 actor: *actor,
443 },
444 }
445}
446
447fn action_from_mutation(mutation: ProfileMutation) -> ProfileAction {
448 match mutation {
449 ProfileMutation::Create { owner, profile } => ProfileAction::Create { owner, profile },
450 ProfileMutation::CreateNamed {
451 owner,
452 name,
453 profile,
454 } => ProfileAction::CreateNamed {
455 owner,
456 name,
457 profile,
458 },
459 ProfileMutation::CreateNamedColored {
460 owner,
461 name,
462 color,
463 profile,
464 } => ProfileAction::CreateNamedColored {
465 owner,
466 name,
467 color,
468 profile,
469 },
470 ProfileMutation::Rename {
471 profile_id,
472 actor,
473 name,
474 } => ProfileAction::Rename {
475 profile_id,
476 actor,
477 name,
478 },
479 ProfileMutation::SetColor {
480 profile_id,
481 actor,
482 color,
483 } => ProfileAction::SetColor {
484 profile_id,
485 actor,
486 color,
487 },
488 ProfileMutation::Replace {
489 profile_id,
490 actor,
491 profile,
492 } => ProfileAction::Replace {
493 profile_id,
494 actor,
495 profile,
496 },
497 ProfileMutation::Delete { profile_id, actor } => {
498 ProfileAction::Delete { profile_id, actor }
499 }
500 }
501}
502
503fn invalidate_state_for_reorg(state: &mut FacadeState) {
504 if state.fault.is_none() {
505 state.fault = Some("access profile facade invalidated by reorganization".to_owned());
506 }
507 state.pending.clear();
508}
509
510#[derive(Debug, Eq, PartialEq)]
511enum ReconciliationDecision {
512 Outcome(ApplyOutcome),
513 Error(String),
514 Fault(String),
515}
516
517fn reconciliation_decision(
518 expected: &ProfileAction,
519 callback: Option<&CallbackRecord>,
520 submission: &Result<TxId, String>,
521) -> ReconciliationDecision {
522 if let Some(callback) = callback {
523 if &callback.action != expected {
524 return ReconciliationDecision::Fault("profile callback action mismatch".to_owned());
525 }
526 if let Ok(submitted_txid) = submission
527 && *submitted_txid != callback.txid
528 {
529 return ReconciliationDecision::Fault(
530 "profile callback transaction mismatch".to_owned(),
531 );
532 }
533 return ReconciliationDecision::Outcome(callback.outcome.clone());
534 }
535 match submission {
536 Ok(_) => ReconciliationDecision::Fault(
537 "peering succeeded without matching profile callback".to_owned(),
538 ),
539 Err(error) => ReconciliationDecision::Error(error.clone()),
540 }
541}
542
543#[cfg(test)]
544mod tests {
545 use super::*;
546 use super::{PendingOperation as Pending, ProfileAction::*};
547
548 #[test]
549 fn mappings_and_reorganization() {
550 let user = UserId::from_tx_id(TxId::from_bytes([1; 12]));
551 let profile_id = ProfileId::new(TxId::from_bytes([2; 12]));
552 let name = ProfileName::new("Named profile".to_owned()).unwrap();
553 let color = ProfileColor::new("violet".to_owned()).unwrap();
554 let profile =
555 AuthorizationProfile::new(vec![ProfileOwner::RequestUser], Vec::new()).unwrap();
556 let actions = [
557 Create {
558 owner: user,
559 profile: profile.clone(),
560 },
561 CreateNamed {
562 owner: user,
563 name: name.clone(),
564 profile: profile.clone(),
565 },
566 CreateNamedColored {
567 owner: user,
568 name: name.clone(),
569 color: Some(color.clone()),
570 profile: profile.clone(),
571 },
572 Replace {
573 profile_id,
574 actor: user,
575 profile,
576 },
577 Rename {
578 profile_id,
579 actor: user,
580 name,
581 },
582 SetColor {
583 profile_id,
584 actor: user,
585 color: Some(color),
586 },
587 Delete {
588 profile_id,
589 actor: user,
590 },
591 ];
592 let mut state = FacadeState::default();
593 for (i, action) in actions.into_iter().enumerate() {
594 assert_eq!(action_from_mutation(mutation_from_action(&action)), action);
595 state.pending.insert(
596 [i as u8; 16],
597 Pending {
598 expected: action,
599 callback: None,
600 },
601 );
602 }
603 invalidate_state_for_reorg(&mut state);
604 assert!(state.fault.is_some() && state.pending.is_empty());
605 }
606}