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