kcode-k1-groups-projection 0.2.0

use std::{
    path::Path,
    sync::{
        Mutex, MutexGuard, RwLock, RwLockReadGuard, RwLockWriteGuard,
        atomic::{AtomicBool, Ordering},
    },
    time::{Duration, Instant},
};

pub use kcode_k1_groups_domain::{
    ALL_MODELS, ALL_USERS, ApplyOutcome, Group, GroupAction, GroupId, GroupMemberships, GroupName,
    GroupRevision, GroupRole, GroupUser, LOCAL_MODELS, ModelId, SentinelGroup, UserId,
};
use kcode_k1_groups_sqlite::Store;
use kcode_k1_groups_state::State;
use kcode_k1_txn_ordering::K1TxnOrdering;
pub use kcode_k1_txn_ordering::TxId;

pub struct Projection {
    state: RwLock<State>,
    apply_lane: Mutex<()>,
    store: Mutex<Store>,
    unavailable: AtomicBool,
}

impl Projection {
    pub fn open(root: &Path, ordering: &K1TxnOrdering) -> Result<(Self, Option<TxId>), String> {
        let started = Instant::now();
        let result = Self::open_inner(root, ordering);
        let elapsed = started.elapsed();
        if elapsed > Duration::from_millis(100) {
            let outcome = if result.is_ok() { "ready" } else { "error" };
            eprintln!(
                "level=warn module=kcode-k1-groups-projection operation=open elapsed_us={} outcome={outcome}",
                elapsed.as_micros()
            );
        }
        result
    }

    fn open_inner(root: &Path, ordering: &K1TxnOrdering) -> Result<(Self, Option<TxId>), String> {
        let opened = kcode_k1_groups_recovery::open(root, ordering)?;
        let (store, state) = opened.into_parts();
        let cursor = state.cursor();
        Ok((
            Self {
                state: RwLock::new(state),
                apply_lane: Mutex::new(()),
                store: Mutex::new(store),
                unavailable: AtomicBool::new(false),
            },
            cursor,
        ))
    }

    pub fn apply(&self, callback_txid: TxId, action: GroupAction) -> Result<ApplyOutcome, String> {
        self.ensure_available()?;
        let _apply_lane = self.lock_apply_lane()?;
        self.ensure_available()?;
        let prepared = {
            let mut state = self.lock_state_write()?;
            state.prepare(callback_txid, action)?
        };
        let receipt = {
            let mut store = self.lock_store()?;
            match store.commit(callback_txid, prepared.change()) {
                Ok(receipt) => receipt,
                Err(error) => {
                    self.mark_unavailable();
                    return Err(error);
                }
            }
        };
        let published = {
            let mut state = self.lock_state_write()?;
            state.publish(callback_txid, prepared, &receipt)
        };
        if published.is_err() {
            self.mark_unavailable();
        }
        published
    }

    pub fn get(&self, group: GroupId) -> Result<Option<Group>, String> {
        self.ensure_available()?;
        let state = self.lock_state_read()?;
        self.ensure_available()?;
        state.get(group)
    }

    pub fn groups_for_user(&self, user: UserId) -> Result<Vec<GroupId>, String> {
        self.ensure_available()?;
        let state = self.lock_state_read()?;
        self.ensure_available()?;
        state.groups_for_user(user)
    }

    pub fn groups_for_model(&self, model: ModelId) -> Result<Vec<GroupId>, String> {
        self.ensure_available()?;
        let state = self.lock_state_read()?;
        self.ensure_available()?;
        state.groups_for_model(model)
    }

    pub fn memberships(&self, user: UserId, model: ModelId) -> Result<GroupMemberships, String> {
        self.ensure_available()?;
        let state = self.lock_state_read()?;
        self.ensure_available()?;
        state.memberships(user, model)
    }

    pub fn clear(&self) -> Result<(), String> {
        self.ensure_available()?;
        let _apply_lane = self.lock_apply_lane()?;
        self.ensure_available()?;
        let receipt = {
            let mut store = self.lock_store()?;
            match store.clear() {
                Ok(receipt) => receipt,
                Err(error) => {
                    self.mark_unavailable();
                    return Err(error);
                }
            }
        };
        let mut state = self.lock_state_write()?;
        state.clear_after_persist(&receipt);
        Ok(())
    }

    fn ensure_available(&self) -> Result<(), String> {
        if self.unavailable.load(Ordering::SeqCst) {
            Err("projection is unavailable until reopen".to_owned())
        } else {
            Ok(())
        }
    }

    fn mark_unavailable(&self) {
        self.unavailable.store(true, Ordering::SeqCst);
    }

    fn lock_apply_lane(&self) -> Result<MutexGuard<'_, ()>, String> {
        self.apply_lane.lock().map_err(|_| {
            self.mark_unavailable();
            "projection apply lane is unavailable until reopen".to_owned()
        })
    }

    fn lock_store(&self) -> Result<MutexGuard<'_, Store>, String> {
        self.store.lock().map_err(|_| {
            self.mark_unavailable();
            "projection database lane is unavailable until reopen".to_owned()
        })
    }

    fn lock_state_read(&self) -> Result<RwLockReadGuard<'_, State>, String> {
        self.state.read().map_err(|_| {
            self.mark_unavailable();
            "projection state is unavailable until reopen".to_owned()
        })
    }

    fn lock_state_write(&self) -> Result<RwLockWriteGuard<'_, State>, String> {
        self.state.write().map_err(|_| {
            self.mark_unavailable();
            "projection state is unavailable until reopen".to_owned()
        })
    }
}

#[cfg(test)]
mod tests {
    use super::Projection;

    fn assert_send_sync<T: Send + Sync>() {}

    #[test]
    fn projection_is_send_and_sync() {
        assert_send_sync::<Projection>();
    }
}