1use std::{
2 collections::HashMap,
3 path::Path,
4 sync::{LockResult, Mutex, RwLock, RwLockReadGuard},
5};
6
7pub use kcode_k1_access_format::{AccessAction, OwnerWitness};
8pub use kcode_k1_access_types::{
9 AccessCheck, AccessId, AccessRevision, Authorizations, GroupId, ModelId, OwnerSubject,
10 RequestPrincipal, SubsystemId, Target, TxId, UserId, ViewerSubject,
11};
12pub use kcode_k1_txn_ordering::K1TxnOrdering;
13
14use kcode_k1_access_discovery_store::DiscoveryStore;
15use kcode_k1_access_store::{Store, StoreMutation, StoredAccess};
16use kcode_k1_transaction::Transaction;
17
18#[derive(Clone, Debug, Eq, PartialEq)]
19pub enum ApplyOutcome {
20 Applied(AccessRevision),
21 Unchanged(AccessRevision),
22 Rejected(String),
23}
24
25pub struct Projection {
26 store: Store,
27 discovery: DiscoveryStore,
28 state: RwLock<State>,
29 apply: Mutex<()>,
30}
31
32struct State {
33 available: bool,
34 objects: HashMap<AccessId, StoredAccess>,
35 targets: HashMap<Target, AccessId>,
36}
37
38struct Prepared {
39 outcome: ApplyOutcome,
40 mutation: StoreMutation,
41 delta: Delta,
42 discovery: Option<(AccessId, Authorizations)>,
43}
44
45enum Delta {
46 None,
47 Create(AccessId, Target, StoredAccess),
48 Replace(AccessId, StoredAccess),
49}
50
51fn locked<T>(result: LockResult<T>, message: &str) -> Result<T, String> {
52 result.map_err(|_| message.to_owned())
53}
54
55fn subjects(authorizations: &Authorizations) -> (Vec<UserId>, Vec<GroupId>) {
56 let mut users = Vec::new();
57 let mut groups = Vec::new();
58 for owner in authorizations.owners() {
59 match owner {
60 OwnerSubject::User(user) => users.push(*user),
61 OwnerSubject::Group(group) => groups.push(*group),
62 }
63 }
64 for viewer in authorizations.viewers() {
65 match viewer {
66 ViewerSubject::User(user) => users.push(*user),
67 ViewerSubject::Group(group) => groups.push(*group),
68 ViewerSubject::Model(_) => {}
69 }
70 }
71 (users, groups)
72}
73
74impl State {
75 fn finish(&mut self, delta: Delta) -> Result<(), String> {
76 let failure = match delta {
77 Delta::None => None,
78 Delta::Create(access_id, target, stored) => {
79 if self.objects.contains_key(&access_id) || self.targets.contains_key(&target) {
80 Some("projection create contradiction after commit")
81 } else {
82 self.targets.insert(target, access_id);
83 self.objects.insert(access_id, stored);
84 None
85 }
86 }
87 Delta::Replace(access_id, stored) => {
88 let consistent = self.targets.get(stored.target()) == Some(&access_id)
89 && self
90 .objects
91 .get(&access_id)
92 .is_some_and(|row| row.target() == stored.target());
93 if !consistent {
94 Some("projection replace contradiction after commit")
95 } else if let Some(row) = self.objects.get_mut(&access_id) {
96 *row = stored;
97 None
98 } else {
99 Some("projection replace disappeared after commit")
100 }
101 }
102 };
103 if let Some(failure) = failure {
104 self.available = false;
105 return Err(failure.to_owned());
106 }
107 Ok(())
108 }
109}
110
111impl Projection {
112 pub fn open(root: &Path, ordering: &K1TxnOrdering) -> Result<(Self, Option<TxId>), String> {
113 let store = Store::open(root)?;
114 let discovery = DiscoveryStore::open(root)?;
115 let (cursor, rows) = store.snapshot()?.into_parts();
116 let subsystem = SubsystemId::from_str("k1-access-subsystem")?;
117 if !Self::valid_snapshot(ordering, subsystem, cursor, &rows)?
118 || discovery.checkpoint()? != cursor
119 {
120 return Self::reset(store, discovery, root);
121 }
122 for row in &rows {
123 let (users, groups) = subjects(row.authorizations());
124 if discovery.contains_missing(&users, &groups, row.access_id())? {
125 return Self::reset(store, discovery, root);
126 }
127 }
128 let mut objects = HashMap::new();
129 let mut targets = HashMap::new();
130 objects
131 .try_reserve(rows.len())
132 .map_err(|error| format!("unable to reserve access index: {error}"))?;
133 targets
134 .try_reserve(rows.len())
135 .map_err(|error| format!("unable to reserve target index: {error}"))?;
136 for row in rows {
137 let access_id = row.access_id();
138 let target = row.target().clone();
139 if objects.insert(access_id, row).is_some()
140 || targets.insert(target, access_id).is_some()
141 {
142 return Self::reset(store, discovery, root);
143 }
144 }
145 Ok((Self::new(store, discovery, objects, targets), cursor))
146 }
147
148 fn new(
149 store: Store,
150 discovery: DiscoveryStore,
151 objects: HashMap<AccessId, StoredAccess>,
152 targets: HashMap<Target, AccessId>,
153 ) -> Self {
154 Self {
155 store,
156 discovery,
157 state: RwLock::new(State {
158 available: true,
159 objects,
160 targets,
161 }),
162 apply: Mutex::new(()),
163 }
164 }
165
166 fn reset(
167 store: Store,
168 discovery: DiscoveryStore,
169 root: &Path,
170 ) -> Result<(Self, Option<TxId>), String> {
171 store.clear()?;
172 discovery.discard()?;
173 let discovery = DiscoveryStore::open(root)?;
174 Ok((
175 Self::new(store, discovery, HashMap::new(), HashMap::new()),
176 None,
177 ))
178 }
179
180 fn valid_snapshot(
181 ordering: &K1TxnOrdering,
182 subsystem: SubsystemId,
183 cursor: Option<TxId>,
184 rows: &[StoredAccess],
185 ) -> Result<bool, String> {
186 if !rows.is_empty() && cursor.is_none() {
187 return Ok(false);
188 }
189 if let Some(cursor) = cursor
190 && Self::load_action(ordering, subsystem, cursor)?.is_none()
191 {
192 return Ok(false);
193 }
194 for row in rows {
195 let access_id = row.access_id();
196 let Some(create) = Self::load_action(ordering, subsystem, access_id.txid())? else {
197 return Ok(false);
198 };
199 let valid = match create {
200 AccessAction::Create {
201 target,
202 authorizations,
203 } if &target == row.target() => {
204 if row.revision() == access_id.txid() {
205 &authorizations == row.authorizations()
206 } else {
207 matches!(Self::load_action(ordering, subsystem, row.revision())?, Some(AccessAction::Replace { access_id: replaced, authorizations, .. }) if replaced == access_id && &authorizations == row.authorizations())
208 }
209 }
210 _ => false,
211 };
212 if !valid {
213 return Ok(false);
214 }
215 }
216 Ok(true)
217 }
218
219 fn load_action(
220 ordering: &K1TxnOrdering,
221 subsystem: SubsystemId,
222 txid: TxId,
223 ) -> Result<Option<AccessAction>, String> {
224 let Some(bytes) = ordering
225 .get_txn(txid)
226 .map_err(|error| format!("KTO transaction lookup failed: {error}"))?
227 else {
228 return Ok(None);
229 };
230 let Ok(transaction) = Transaction::parse(&bytes) else {
231 return Ok(None);
232 };
233 if transaction.subsystem() != subsystem {
234 return Ok(None);
235 }
236 Ok(kcode_k1_access_format::decode(transaction.payload())
237 .ok()
238 .map(|(_, action)| action))
239 }
240
241 pub fn apply(&self, callback_txid: TxId, action: AccessAction) -> Result<ApplyOutcome, String> {
242 let _lane = locked(self.apply.lock(), "projection apply lock poisoned")?;
243 let prepared = {
244 let mut state = locked(self.state.write(), "projection state lock poisoned")?;
245 if !state.available {
246 return Err("projection unavailable".to_owned());
247 }
248 Self::prepare(&mut state, callback_txid, action)?
249 };
250 if let Err(error) = self.store.commit(callback_txid, &prepared.mutation) {
251 self.unavailable();
252 return Err(error);
253 }
254 let committed = if let Some((access_id, authorizations)) = &prepared.discovery {
255 let (users, groups) = subjects(authorizations);
256 self.discovery
257 .commit(callback_txid, &users, &groups, *access_id)
258 } else {
259 self.discovery
260 .commit(callback_txid, &[], &[], AccessId::new(callback_txid))
261 };
262 if let Err(error) = committed {
263 self.unavailable();
264 return Err(error);
265 }
266 let mut state = locked(
267 self.state.write(),
268 "projection state lock poisoned after commit",
269 )?;
270 if !state.available {
271 return Err("projection became unavailable after commit".to_owned());
272 }
273 state.finish(prepared.delta)?;
274 Ok(prepared.outcome)
275 }
276
277 fn unavailable(&self) {
278 if let Ok(mut state) = self.state.write() {
279 state.available = false;
280 }
281 }
282
283 fn prepare(state: &mut State, txid: TxId, action: AccessAction) -> Result<Prepared, String> {
284 match action {
285 AccessAction::Create {
286 target,
287 authorizations,
288 } => {
289 let access_id = AccessId::new(txid);
290 if state.objects.contains_key(&access_id) {
291 return Ok(Self::rejected("access ID already exists"));
292 }
293 if state.targets.contains_key(&target) {
294 return Ok(Self::rejected("target already exists"));
295 }
296 state
297 .objects
298 .try_reserve(1)
299 .map_err(|error| format!("unable to reserve access index: {error}"))?;
300 state
301 .targets
302 .try_reserve(1)
303 .map_err(|error| format!("unable to reserve target index: {error}"))?;
304 let stored =
305 StoredAccess::new(access_id, target.clone(), txid, authorizations.clone());
306 Ok(Prepared {
307 outcome: ApplyOutcome::Applied(AccessRevision::new(access_id, txid)),
308 mutation: StoreMutation::Create(stored.clone()),
309 delta: Delta::Create(access_id, target, stored),
310 discovery: Some((access_id, authorizations)),
311 })
312 }
313 AccessAction::Replace {
314 access_id,
315 actor,
316 witness,
317 authorizations,
318 ..
319 } => {
320 let Some(current) = state.objects.get(&access_id) else {
321 return Ok(Self::rejected("unknown access ID"));
322 };
323 if !Self::valid_witness(current.authorizations(), actor, &witness) {
324 return Ok(Self::rejected("owner witness is invalid"));
325 }
326 if current.authorizations() == &authorizations {
327 return Ok(Prepared {
328 outcome: ApplyOutcome::Unchanged(AccessRevision::new(
329 access_id,
330 current.revision(),
331 )),
332 mutation: StoreMutation::CursorOnly,
333 delta: Delta::None,
334 discovery: Some((access_id, authorizations)),
335 });
336 }
337 let stored = StoredAccess::new(
338 access_id,
339 current.target().clone(),
340 txid,
341 authorizations.clone(),
342 );
343 Ok(Prepared {
344 outcome: ApplyOutcome::Applied(AccessRevision::new(access_id, txid)),
345 mutation: StoreMutation::Replace {
346 access_id,
347 revision: txid,
348 authorizations: authorizations.clone(),
349 },
350 delta: Delta::Replace(access_id, stored),
351 discovery: Some((access_id, authorizations)),
352 })
353 }
354 AccessAction::EnsureDiscovery { access_id } => {
355 let Some(stored) = state.objects.get(&access_id) else {
356 return Ok(Self::rejected("unknown access ID"));
357 };
358 Ok(Prepared {
359 outcome: ApplyOutcome::Unchanged(AccessRevision::new(
360 access_id,
361 stored.revision(),
362 )),
363 mutation: StoreMutation::CursorOnly,
364 delta: Delta::None,
365 discovery: Some((access_id, stored.authorizations().clone())),
366 })
367 }
368 }
369 }
370
371 fn rejected(reason: &str) -> Prepared {
372 Prepared {
373 outcome: ApplyOutcome::Rejected(reason.to_owned()),
374 mutation: StoreMutation::CursorOnly,
375 delta: Delta::None,
376 discovery: None,
377 }
378 }
379
380 fn valid_witness(auth: &Authorizations, actor: UserId, witness: &OwnerWitness) -> bool {
381 auth.owners().iter().any(|owner| match (owner, witness) {
382 (OwnerSubject::User(owner), OwnerWitness::User) => *owner == actor,
383 (OwnerSubject::Group(owner), OwnerWitness::Group(group)) => owner == group,
384 _ => false,
385 })
386 }
387
388 fn readable(&self) -> Result<RwLockReadGuard<'_, State>, String> {
389 let state = locked(self.state.read(), "projection state lock poisoned")?;
390 if state.available {
391 Ok(state)
392 } else {
393 Err("projection unavailable".to_owned())
394 }
395 }
396
397 pub fn discovery_missing(
398 &self,
399 access_id: AccessId,
400 expected_subsystem: SubsystemId,
401 ) -> Result<bool, String> {
402 let state = self.readable()?;
403 let Some(stored) = state.objects.get(&access_id) else {
404 return Ok(false);
405 };
406 if stored.target().subsystem() != expected_subsystem {
407 return Ok(false);
408 }
409 let (users, groups) = subjects(stored.authorizations());
410 self.discovery.contains_missing(&users, &groups, access_id)
411 }
412
413 pub fn discovered_for_user(
414 &self,
415 user: UserId,
416 subsystem: SubsystemId,
417 ) -> Result<Vec<AccessId>, String> {
418 self.discovered(self.discovery.list_user(user)?, subsystem)
419 }
420
421 pub fn discovered_for_group(
422 &self,
423 group: GroupId,
424 subsystem: SubsystemId,
425 ) -> Result<Vec<AccessId>, String> {
426 self.discovered(self.discovery.list_group(group)?, subsystem)
427 }
428
429 fn discovered(
430 &self,
431 ids: Vec<AccessId>,
432 subsystem: SubsystemId,
433 ) -> Result<Vec<AccessId>, String> {
434 let state = self.readable()?;
435 Ok(ids
436 .into_iter()
437 .filter(|id| {
438 state
439 .objects
440 .get(id)
441 .is_some_and(|stored| stored.target().subsystem() == subsystem)
442 })
443 .collect())
444 }
445
446 pub fn owner_witness(
447 &self,
448 access_id: AccessId,
449 user: UserId,
450 user_groups: &[GroupId],
451 ) -> Result<Option<OwnerWitness>, String> {
452 let state = self.readable()?;
453 let Some(stored) = state.objects.get(&access_id) else {
454 return Ok(None);
455 };
456 let owners = stored.authorizations().owners();
457 if owners
458 .iter()
459 .any(|owner| matches!(owner, OwnerSubject::User(owner) if *owner == user))
460 {
461 return Ok(Some(OwnerWitness::User));
462 }
463 Ok(owners
464 .iter()
465 .filter_map(|owner| match owner {
466 OwnerSubject::Group(group) if user_groups.contains(group) => Some(*group),
467 _ => None,
468 })
469 .min()
470 .map(OwnerWitness::Group))
471 }
472
473 pub fn check(
474 &self,
475 principal: RequestPrincipal,
476 access_id: AccessId,
477 expected_subsystem: SubsystemId,
478 user_groups: &[GroupId],
479 model_groups: &[GroupId],
480 groups_revision: Option<TxId>,
481 ) -> Result<AccessCheck, String> {
482 let state = self.readable()?;
483 let Some(stored) = state.objects.get(&access_id) else {
484 return Self::hidden_check();
485 };
486 if stored.target().subsystem() != expected_subsystem {
487 return Self::hidden_check();
488 }
489 let auth = stored.authorizations();
490 let user_owner = Self::user_owner(auth, principal.user(), user_groups);
491 let user_view = user_owner
492 || auth.viewers().iter().any(|viewer| match viewer {
493 ViewerSubject::User(user) => *user == principal.user(),
494 ViewerSubject::Group(group) => user_groups.contains(group),
495 ViewerSubject::Model(_) => false,
496 });
497 let model_view = auth.owners().iter().any(|owner| match owner {
498 OwnerSubject::Group(group) => model_groups.contains(group),
499 OwnerSubject::User(_) => false,
500 }) || auth.viewers().iter().any(|viewer| match viewer {
501 ViewerSubject::Model(model) => *model == principal.model(),
502 ViewerSubject::Group(group) => model_groups.contains(group),
503 ViewerSubject::User(_) => false,
504 });
505 let can_manage = user_owner;
506 let can_view = user_view && model_view;
507 let evidence = can_view || can_manage;
508 AccessCheck::new(
509 can_view,
510 can_manage,
511 can_view.then(|| stored.target().clone()),
512 evidence.then_some(stored.revision()),
513 evidence.then_some(groups_revision).flatten(),
514 )
515 }
516
517 fn user_owner(auth: &Authorizations, user: UserId, groups: &[GroupId]) -> bool {
518 auth.owners().iter().any(|owner| match owner {
519 OwnerSubject::User(owner) => *owner == user,
520 OwnerSubject::Group(group) => groups.contains(group),
521 })
522 }
523
524 fn hidden_check() -> Result<AccessCheck, String> {
525 AccessCheck::new(false, false, None, None, None)
526 }
527
528 pub fn clear(&self) -> Result<(), String> {
529 let _lane = locked(self.apply.lock(), "projection apply lock poisoned")?;
530 self.unavailable();
531 self.store.clear().and(self.discovery.discard())
532 }
533}
534
535#[cfg(test)]
536mod tests {
537 use super::*;
538 use std::{
539 fs,
540 sync::atomic::{AtomicU64, Ordering},
541 };
542
543 static NEXT: AtomicU64 = AtomicU64::new(0);
544
545 fn root() -> std::path::PathBuf {
546 let root = std::env::temp_dir().join(format!(
547 "access-projection-{}-{}",
548 std::process::id(),
549 NEXT.fetch_add(1, Ordering::Relaxed)
550 ));
551 fs::create_dir(&root).unwrap();
552 root
553 }
554
555 fn tx(value: u8) -> TxId {
556 TxId::from_bytes([value; 12])
557 }
558 fn user(value: u8) -> UserId {
559 UserId::from_tx_id(tx(value))
560 }
561 fn group(value: u8) -> GroupId {
562 GroupId::new(tx(value))
563 }
564 fn target(name: &str) -> Target {
565 Target::new(SubsystemId::from_str(name).unwrap(), vec![1])
566 }
567 fn auth() -> Authorizations {
568 Authorizations::new(
569 vec![OwnerSubject::User(user(1)), OwnerSubject::Group(group(2))],
570 vec![
571 ViewerSubject::User(user(3)),
572 ViewerSubject::Group(group(4)),
573 ViewerSubject::Model(ModelId::from_bytes([5; 32])),
574 ],
575 )
576 .unwrap()
577 }
578
579 #[test]
580 fn create_replace_ensure_queries_and_authorization_fan_out_without_duplicates()
581 -> Result<(), String> {
582 let root = root();
583 let ordering = K1TxnOrdering::open(&root.join("ordering"))?;
584 let (projection, _) = Projection::open(&root.join("projection"), &ordering)?;
585 let access_id = AccessId::new(tx(6));
586 let subsystem = SubsystemId::from_str("one")?;
587 assert_eq!(
588 projection.apply(
589 tx(6),
590 AccessAction::Create {
591 target: target("one"),
592 authorizations: auth()
593 }
594 )?,
595 ApplyOutcome::Applied(AccessRevision::new(access_id, tx(6)))
596 );
597 assert_eq!(
598 projection.discovered_for_user(user(1), subsystem)?,
599 vec![access_id]
600 );
601 assert_eq!(
602 projection.discovered_for_group(group(4), subsystem)?,
603 vec![access_id]
604 );
605 assert!(!projection.discovery_missing(access_id, subsystem)?);
606 assert_eq!(
607 projection.apply(tx(7), AccessAction::EnsureDiscovery { access_id })?,
608 ApplyOutcome::Unchanged(AccessRevision::new(access_id, tx(6)))
609 );
610 assert_eq!(
611 projection.discovered_for_user(user(1), subsystem)?,
612 vec![access_id]
613 );
614 let replacement = Authorizations::new(
615 vec![OwnerSubject::User(user(3))],
616 vec![ViewerSubject::Group(group(2))],
617 )?;
618 assert_eq!(
619 projection.apply(
620 tx(8),
621 AccessAction::Replace {
622 access_id,
623 actor: user(1),
624 groups_revision: None,
625 witness: OwnerWitness::User,
626 authorizations: replacement
627 }
628 )?,
629 ApplyOutcome::Applied(AccessRevision::new(access_id, tx(8)))
630 );
631 assert_eq!(
632 projection.discovered_for_user(user(3), subsystem)?,
633 vec![access_id]
634 );
635 assert_eq!(
636 projection.discovered_for_group(group(2), subsystem)?,
637 vec![access_id]
638 );
639 assert!(
640 projection
641 .discovered_for_user(user(1), SubsystemId::from_str("two")?)?
642 .is_empty()
643 );
644 assert_eq!(
645 projection.apply(
646 tx(9),
647 AccessAction::EnsureDiscovery {
648 access_id: AccessId::new(tx(9))
649 }
650 )?,
651 ApplyOutcome::Rejected("unknown access ID".to_owned())
652 );
653 let view = projection.check(
654 RequestPrincipal::new(user(1), ModelId::from_bytes([5; 32])),
655 access_id,
656 subsystem,
657 &[group(2)],
658 &[group(2)],
659 None,
660 )?;
661 assert!(view.can_view() && !view.can_manage());
662 assert_eq!(
663 projection.owner_witness(access_id, user(3), &[group(2)])?,
664 Some(OwnerWitness::User)
665 );
666 fs::remove_dir_all(root).map_err(|error| error.to_string())
667 }
668
669 #[test]
670 fn mismatched_discovery_checkpoint_resets_both_derived_stores() -> Result<(), String> {
671 let root = root();
672 let projection_root = root.join("projection");
673 let ordering = K1TxnOrdering::open(&root.join("ordering"))?;
674 kcode_k1_access_discovery_store::DiscoveryStore::open(&projection_root)?.commit(
675 tx(1),
676 &[user(2)],
677 &[],
678 AccessId::new(tx(3)),
679 )?;
680 let (projection, cursor) = Projection::open(&projection_root, &ordering)?;
681 assert_eq!(cursor, None);
682 assert!(
683 projection
684 .discovered_for_user(user(2), SubsystemId::from_str("one")?)?
685 .is_empty()
686 );
687 projection.clear()?;
688 fs::remove_dir_all(root).map_err(|error| error.to_string())
689 }
690}