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