kcode_k1_groups_projection/
lib.rs

1use std::{
2    path::Path,
3    sync::{
4        Mutex, MutexGuard, RwLock, RwLockReadGuard, RwLockWriteGuard,
5        atomic::{AtomicBool, Ordering},
6    },
7    time::{Duration, Instant},
8};
9
10pub use kcode_k1_groups_domain::{
11    ALL_MODELS, ALL_USERS, ApplyOutcome, Group, GroupAction, GroupId, GroupMemberships, GroupName,
12    GroupRevision, GroupRole, GroupUser, LOCAL_MODELS, ModelId, SentinelGroup, UserId,
13};
14use kcode_k1_groups_sqlite::Store;
15use kcode_k1_groups_state::State;
16use kcode_k1_txn_ordering::K1TxnOrdering;
17pub use kcode_k1_txn_ordering::TxId;
18
19pub struct Projection {
20    state: RwLock<State>,
21    apply_lane: Mutex<()>,
22    store: Mutex<Store>,
23    unavailable: AtomicBool,
24}
25
26impl Projection {
27    pub fn open(root: &Path, ordering: &K1TxnOrdering) -> Result<(Self, Option<TxId>), String> {
28        let started = Instant::now();
29        let result = Self::open_inner(root, ordering);
30        let elapsed = started.elapsed();
31        if elapsed > Duration::from_millis(100) {
32            let outcome = if result.is_ok() { "ready" } else { "error" };
33            eprintln!(
34                "level=warn module=kcode-k1-groups-projection operation=open elapsed_us={} outcome={outcome}",
35                elapsed.as_micros()
36            );
37        }
38        result
39    }
40
41    fn open_inner(root: &Path, ordering: &K1TxnOrdering) -> Result<(Self, Option<TxId>), String> {
42        let opened = kcode_k1_groups_recovery::open(root, ordering)?;
43        let (store, state) = opened.into_parts();
44        let cursor = state.cursor();
45        Ok((
46            Self {
47                state: RwLock::new(state),
48                apply_lane: Mutex::new(()),
49                store: Mutex::new(store),
50                unavailable: AtomicBool::new(false),
51            },
52            cursor,
53        ))
54    }
55
56    pub fn apply(&self, callback_txid: TxId, action: GroupAction) -> Result<ApplyOutcome, String> {
57        self.ensure_available()?;
58        let _apply_lane = self.lock_apply_lane()?;
59        self.ensure_available()?;
60        let prepared = {
61            let mut state = self.lock_state_write()?;
62            state.prepare(callback_txid, action)?
63        };
64        let receipt = {
65            let mut store = self.lock_store()?;
66            match store.commit(callback_txid, prepared.change()) {
67                Ok(receipt) => receipt,
68                Err(error) => {
69                    self.mark_unavailable();
70                    return Err(error);
71                }
72            }
73        };
74        let published = {
75            let mut state = self.lock_state_write()?;
76            state.publish(callback_txid, prepared, &receipt)
77        };
78        if published.is_err() {
79            self.mark_unavailable();
80        }
81        published
82    }
83
84    pub fn get(&self, group: GroupId) -> Result<Option<Group>, String> {
85        self.ensure_available()?;
86        let state = self.lock_state_read()?;
87        self.ensure_available()?;
88        state.get(group)
89    }
90
91    pub fn groups_for_user(&self, user: UserId) -> Result<Vec<GroupId>, String> {
92        self.ensure_available()?;
93        let state = self.lock_state_read()?;
94        self.ensure_available()?;
95        state.groups_for_user(user)
96    }
97
98    pub fn groups_for_model(&self, model: ModelId) -> Result<Vec<GroupId>, String> {
99        self.ensure_available()?;
100        let state = self.lock_state_read()?;
101        self.ensure_available()?;
102        state.groups_for_model(model)
103    }
104
105    pub fn memberships(&self, user: UserId, model: ModelId) -> Result<GroupMemberships, String> {
106        self.ensure_available()?;
107        let state = self.lock_state_read()?;
108        self.ensure_available()?;
109        state.memberships(user, model)
110    }
111
112    pub fn clear(&self) -> Result<(), String> {
113        self.ensure_available()?;
114        let _apply_lane = self.lock_apply_lane()?;
115        self.ensure_available()?;
116        let receipt = {
117            let mut store = self.lock_store()?;
118            match store.clear() {
119                Ok(receipt) => receipt,
120                Err(error) => {
121                    self.mark_unavailable();
122                    return Err(error);
123                }
124            }
125        };
126        let mut state = self.lock_state_write()?;
127        state.clear_after_persist(&receipt);
128        Ok(())
129    }
130
131    fn ensure_available(&self) -> Result<(), String> {
132        if self.unavailable.load(Ordering::SeqCst) {
133            Err("projection is unavailable until reopen".to_owned())
134        } else {
135            Ok(())
136        }
137    }
138
139    fn mark_unavailable(&self) {
140        self.unavailable.store(true, Ordering::SeqCst);
141    }
142
143    fn lock_apply_lane(&self) -> Result<MutexGuard<'_, ()>, String> {
144        self.apply_lane.lock().map_err(|_| {
145            self.mark_unavailable();
146            "projection apply lane is unavailable until reopen".to_owned()
147        })
148    }
149
150    fn lock_store(&self) -> Result<MutexGuard<'_, Store>, String> {
151        self.store.lock().map_err(|_| {
152            self.mark_unavailable();
153            "projection database lane is unavailable until reopen".to_owned()
154        })
155    }
156
157    fn lock_state_read(&self) -> Result<RwLockReadGuard<'_, State>, String> {
158        self.state.read().map_err(|_| {
159            self.mark_unavailable();
160            "projection state is unavailable until reopen".to_owned()
161        })
162    }
163
164    fn lock_state_write(&self) -> Result<RwLockWriteGuard<'_, State>, String> {
165        self.state.write().map_err(|_| {
166            self.mark_unavailable();
167            "projection state is unavailable until reopen".to_owned()
168        })
169    }
170}
171
172#[cfg(test)]
173mod tests {
174    use super::Projection;
175
176    fn assert_send_sync<T: Send + Sync>() {}
177
178    #[test]
179    fn projection_is_send_and_sync() {
180        assert_send_sync::<Projection>();
181    }
182}
kcode_k1_groups_projection/lib.rs

kcode_k1_groups_projection/
lib.rs
