crew_rs/

crew.rs

use std::{collections::{hash_map::Entry, HashMap, HashSet}, rc::Rc};

use audi::{Listener, ListenerSet};
use endr::{SetItem, SetItemID, WriteAccess};
use futures::{future, stream, FutureExt, Stream, StreamExt};
use litl::{impl_debug_as_litl, impl_nested_tagged_data_serde, NestedTaggedData};
use mofo::Mofo;
use objt::objt;
use ridl::{
    signing::{SignatureError, Signed},
    symm_encr::KeySecret,
};
use serde::Serialize;
use serde_derive::{Deserialize, Serialize};
use ti64::MsSinceEpoch;

use crate::{
    credential_source::CredentialSource,
    crew_manager::WeakCrewManager,
    timeline::{Timeline, UncheckedTimeline, UncheckedTimelineEntry},
    AddRole, CrewChange, CrewChangeTx, CrewManager, CrewRulesV1, CrewState, MemberCredential,
    RevealSecret, ADMIN_INVITATION_ROLE, ADMIN_ROLE, CONTENT_SECRET, PARTICIPATION_SECRET,
    READER_ROLE, READ_INVITATION_ROLE, SET_WRITE_ACCESS_INFO_ID, WRITER_ROLE,
    WRITE_INVITATION_ROLE,
};

#[derive(Clone, Copy, PartialEq, Eq, Hash)]
pub struct CrewID(pub endr::ObjectID);

impl_debug_as_litl!(CrewID);

impl NestedTaggedData for CrewID {
    const TAG: &'static str = "crew";

    type Inner = endr::ObjectID;

    fn as_inner(&self) -> &Self::Inner {
        &self.0
    }

    fn from_inner(inner: Self::Inner) -> Self
    where
        Self: Sized,
    {
        CrewID(inner)
    }
}

impl_nested_tagged_data_serde!(CrewID);

pub struct CrewInner {
    pub(crate) endr: endr::Node,
    pub(crate) id: CrewID,
    pub(crate) unchecked_timeline: UncheckedTimeline,
    pub(crate) last_update: Option<Rc<CrewUpdate>>,
    pub(crate) credential_source: Box<dyn CredentialSource>,
    pub(crate) listeners: ListenerSet<Rc<CrewUpdate>>,
    pub(crate) manager: CrewManager,
    pub(crate) parents_unchecked: HashMap<CrewID, Crew>,
    pub(crate) background: Mofo,
    pub(crate) shared_secret_cache: HashMap<String, Rc<KeySecret>>,
}

#[derive(Serialize, Deserialize)]
pub struct CrewUpdate {
    pub(crate) timeline: Timeline,
    pub(crate) valid_txs: HashMap<SetItemID, CrewChangeTx>,
    pub(crate) invalid_txs: HashMap<SetItemID, (CrewChangeTx, String)>,
}
impl_debug_as_litl!(CrewUpdate);

impl CrewUpdate {
    pub fn current_state(&self) -> Option<&CrewState> {
        self.timeline.last().map(|(_, state)| state)
    }
}

objt!(
    Crew,
    WeakCrew,
    CrewInner,
    endr: endr::Node,
    listeners: ListenerSet<Rc<CrewUpdate>>
);

impl Crew {
    pub fn id(&self) -> CrewID {
        self.borrow().id
    }
    pub fn current_state(&self) -> Option<CrewState> {
        self.borrow()
            .last_update
            .as_ref()
            .and_then(|u| u.current_state().cloned())
    }
}

impl Crew {
    pub async fn make_changes<I: IntoIterator<Item = CrewChange>>(
        &self,
        changes: I,
    ) -> Result<SetItemID, String> {
        let write_access = litl::from_val::<WriteAccess>(
            self.get_entrusted_info(PARTICIPATION_SECRET, SET_WRITE_ACCESS_INFO_ID)?,
        )
        .map_err(|err| err.to_string())?;

        self.make_changes_with_write_access(changes, write_access)
            .await
    }

    pub(crate) async fn make_changes_with_write_access<I: IntoIterator<Item = CrewChange>>(
        &self,
        changes: I,
        write_access: WriteAccess,
    ) -> Result<SetItemID, String> {
        let set_id = self.borrow().id.0;
        let endr = self.endr();

        let (credential, _, _) = self
            .get_credentials_with_role(&HashSet::from_iter([
                ADMIN_ROLE.to_string(),
                WRITER_ROLE.to_string(),
                ADMIN_INVITATION_ROLE.to_string(),
                WRITE_INVITATION_ROLE.to_string(),
                READ_INVITATION_ROLE.to_owned(),
            ]))
            .first()
            .cloned()
            .expect("Expected credential to make changes");

        let tx = credential.sign_tx(changes);

        let tx_item_id = endr
            .insert_into_set_after_frontier(set_id, &write_access, tx)
            .await
            .map_err(|err| err.to_string())?;

        // wait for tx to pop up as a valid one
        let mut updates = self.updates(format!("Waiting for {:?}", tx_item_id));

        while let Some(update) = updates.next().await {
            if update.valid_txs.contains_key(&tx_item_id) {
                return Ok(tx_item_id);
            } else if let Some(reason) = update
                .invalid_txs
                .get(&tx_item_id)
                .map(|(_, reason)| reason)
            {
                return Err(format!("Tx turned out invalid: {}", reason));
            } else {
                // hasn't appeared yet
                // println!("{} hasn't appeared in update {:#?}", tx_item_id, update);
            }
        }

        Err("Reached end of set updates before finding new tx".to_string())
    }

    pub fn get_credentials_with_role(
        &self,
        preferred_role_first: &HashSet<String>,
    ) -> Vec<(MemberCredential, String, CrewID)> {
        let credential_source = self.borrow().credential_source.clone_ref();
        let all_credentials = credential_source.current_credentials_for(&self.id());
        let mut collected_credentials = Vec::new();

        if let Some(current_state) = self.current_state() {
            let parents = current_state
                .parents
                .iter()
                .map(|parent_id| {
                    self.borrow()
                        .parents_unchecked
                        .get(parent_id)
                        .unwrap()
                        .clone()
                })
                .collect::<Vec<_>>();


            for role in preferred_role_first {
                for credential in &all_credentials {
                    if current_state
                        .roles_of(&credential.signer().pub_id())
                        .contains(role)
                    {
                        collected_credentials.push((credential.clone(), role.clone(), self.id()));
                    }

                }

                for parent in &parents {
                    collected_credentials.extend(
                        parent
                            .get_credentials_with_role(&HashSet::from_iter([role.to_owned()])),
                    );
                }
            }
        }

        collected_credentials
    }

    fn get_parent_for_ancestor(&self, ancestor_id: &CrewID) -> Option<Crew> {
        let own_parents = &self.borrow().parents_unchecked;
        own_parents.get(ancestor_id).cloned().or_else(|| {
            own_parents.values().find_map(|parent| {
                if parent.get_parent_for_ancestor(ancestor_id).is_some() {
                    Some(parent.clone_ref())
                } else {
                    None
                }
            })
        })
    }

    // TODO: super simplifying assumption for now: we only ever get more secrets, one secret per kind
    pub fn get_shared_secret(&self, secret_kind: &str) -> Result<Rc<KeySecret>, String> {
        if let Some(secret) = self.borrow_mut().shared_secret_cache.get(secret_kind) {
            return Ok(Rc::clone(secret));
        }

        let secret = Rc::new(self.get_shared_secret_uncached(secret_kind)?);
        self.borrow_mut().shared_secret_cache.insert(secret_kind.to_owned(), Rc::clone(&secret));
        Ok(secret)
    }

    fn get_shared_secret_uncached(&self, secret_kind: &str) -> Result<KeySecret, String> {
        let credentials = match secret_kind {
            CONTENT_SECRET => self.get_credentials_with_role(&HashSet::from_iter([
                ADMIN_ROLE.to_owned(),
                WRITER_ROLE.to_owned(),
                READER_ROLE.to_owned(),
                ADMIN_INVITATION_ROLE.to_owned(),
                WRITE_INVITATION_ROLE.to_owned(),
                READ_INVITATION_ROLE.to_owned(),
            ])),
            PARTICIPATION_SECRET => self.get_credentials_with_role(&HashSet::from_iter([
                ADMIN_ROLE.to_owned(),
                WRITER_ROLE.to_owned(),
                ADMIN_INVITATION_ROLE.to_owned(),
                WRITE_INVITATION_ROLE.to_owned(),
                READ_INVITATION_ROLE.to_owned(),
            ])),
            _ => self.get_credentials_with_role(&HashSet::from_iter([ADMIN_ROLE.to_owned()])),
        };

        match &self.current_state() {
            Some(state) => {
                let (oks, errs) = credentials
                    .iter()
                    .map(|(credential, credential_role, source_crew)| {
                        if source_crew == &self.id() {
                            let encr_secret = state
                                .shared_secrets
                                .get(secret_kind)
                                .ok_or("No shared secrets for that kind")?
                                .get(&credential.pub_id().signer)
                                .ok_or(format!("Secret not revealed for crew credential {:?} with role {}", credential, credential_role))?;
                            let secret = credential
                                .decrypt_secret(encr_secret)
                                .map_err(|err| err.to_string())?;
                            Ok(secret)
                        } else {
                            let parent_crew = self
                                .get_parent_for_ancestor(&source_crew)
                                .expect("Expected parent crew to be loaded")
                                .clone_ref();
                            let parent_secret = parent_crew.get_shared_secret(secret_kind)?;
                            let entry = state
                                .parent_secret_map
                                .get(&(parent_crew.id(), secret_kind.to_owned(), parent_secret.id))
                                .ok_or_else(|| {
                                    "Expected to have entry in parent secret map".to_owned()
                                })?;
                            let secret = parent_secret
                                .decrypt(entry)
                                .map_err(|err| err.to_string())?;
                            Ok(secret)
                        }
                    })
                    .partition::<Vec<_>, _>(Result::is_ok);
                oks.into_iter().next().unwrap_or_else(|| {
                    Err(format!(
                        "Couldn't find any valid credential for secret kind {}: {:?}",
                        secret_kind, errs
                    ))
                })
            }
            None => Err("No valid state yet".to_string()),
        }
    }

    pub fn get_entrusted_info(
        &self,
        secret_kind: &str,
        info_id: &str,
    ) -> Result<litl::Val, String> {
        let secret = self.get_shared_secret(secret_kind)?;

        match &self.current_state() {
            Some(state) => {
                let encr_info = state
                    .entrusted_info
                    .get(secret_kind)
                    .ok_or("No entrusted infos for that secret kind")?
                    .get(info_id)
                    .ok_or("No such entrusted info")?;
                let info = secret.decrypt(encr_info).map_err(|err| err.to_string())?;
                Ok(info)
            }
            None => Err("No valid state yet".to_string()),
        }
    }

    pub async fn add_update_listener(&self, listener: Listener<Rc<CrewUpdate>>) {
        let initial_update = { self.borrow().last_update.clone() };
        self.listeners()
            .add_with_initial_msg(listener, initial_update)
            .await;
    }

    pub fn updates(
        &self,
        listener_prefix: String,
    ) -> impl Stream<Item = Rc<CrewUpdate>> + Unpin + 'static {
        let (updates_tx, updates_rx) = futures::channel::mpsc::channel(100);

        let self_clone = self.clone();

        stream::once(async move {
            self_clone
                .add_update_listener(Listener::new(
                    &format!("{}_{:?}", listener_prefix, rand::random::<u64>()),
                    updates_tx,
                ))
                .await;

            updates_rx
        })
        .flatten()
        .boxed_local()
    }

    pub async fn handle_set_diff(&self, diff: endr::Diff) {
        let set_diff = diff.expect_set();

        if let Some(header) = &set_diff.header {
            let meta = header.meta.clone().expect("Expected header to have meta");

            assert_eq!(meta.get("type").unwrap(), &litl::Val::str("crew"));

            let initial_state = litl::from_val::<CrewState>(
                meta.get("initialState")
                    .cloned()
                    .expect("Expected initial state"),
            )
            .expect("Expected intial state to deserialize");

            self.borrow_mut()
                .unchecked_timeline
                .set_intial_state(initial_state, MsSinceEpoch(0)); // TODO: better timestamp for header?
        }

        for new_item in &set_diff.new_items {
            let tx = litl::from_val::<CrewChangeTx>(new_item.attested.data.clone())
                .unwrap_or_else(|_| panic!("Expected tx to deserialize {:?}", new_item));

            let mut crew_mut = self.borrow_mut();
            let tx_id = new_item.id();

            crew_mut.unchecked_timeline.insert_own_tx(tx, tx_id);
        }

        let update = self.borrow().unchecked_timeline.resolve(CrewRulesV1);

        let manager = self.borrow().manager.clone();
        for parent_id in update
            .current_state()
            .iter()
            .flat_map(|state| state.parents.iter())
        {
            if self.borrow().parents_unchecked.contains_key(parent_id) {
                continue;
            }
            let parent_crew = manager.load_crew(*parent_id).await;
            self.borrow_mut()
                .parents_unchecked
                .insert(*parent_id, parent_crew.clone());

            let self_bg = self.clone();
            let parent_id = *parent_id;

            self.borrow().background.add_background_task(
                parent_crew
                    .updates("child_crew".to_owned())
                    .for_each(move |update| {
                        let self_bg = self_bg.clone();

                        async move {
                            for (made_at, state) in &update.timeline {
                                self_bg
                                    .borrow_mut()
                                    .unchecked_timeline
                                    .insert_parent_state(state.clone(), parent_id, *made_at)
                            }
                            let update = Rc::new(self_bg.borrow().unchecked_timeline.resolve(CrewRulesV1));
                            self_bg.borrow_mut().last_update = Some(update.clone());
                            self_bg.listeners().broadcast(update).await;
                        }
                    })
                    .boxed_local(),
            )
        }

        let update_rc = Rc::new(update);

        self.borrow_mut().last_update = Some(update_rc.clone());
        self.listeners().broadcast(update_rc).await;
    }

    pub async fn wait_for_state(&self, condition: impl Fn(&CrewState) -> bool) {
        let mut updates = self.updates("waiting for update".to_owned());

        while let Some(update) = updates.next().await {
            if let Some(state) = update.current_state() {
                if condition(state) {
                    return;
                }
            }
        }

        panic!("Reached end of updates stream without matching state")
    }

    pub fn is_signed_by_member_with_role<T: Clone + Serialize>(
        &self,
        signed: &Signed<T>,
        role: &str,
    ) -> Result<(), Vec<SignatureError>> {
        let members_with_role = self
            .current_state()
            .expect("Expected crew state when validating signed data")
            .roles
            .into_iter()
            .filter_map(|(member, member_role)| {
                if member_role == role {
                    Some(member)
                } else {
                    None
                }
            });

        let mut errors = vec![];

        for member in members_with_role {
            match signed.ensure_signed_by(&member.signer) {
                Ok(()) => return Ok(()),
                Err(e) => errors.push(e),
            }
        }

        Err(errors)
    }

    pub async fn create_invitation(&self, invitation_role: &str) -> Result<Invitation, String> {
        let invitation_credential = MemberCredential::new_random();
        let content_secret = self.get_shared_secret(CONTENT_SECRET).map_err(|err| {
            format!(
                "Expected to have content secret when creating invitation: {}",
                err
            )
        })?;
        let participation_secret = self
            .get_shared_secret(PARTICIPATION_SECRET)
            .map_err(|err| {
                format!(
                    "Expected to have participation secret when creating invitation: {}",
                    err
                )
            })?;

        self.make_changes([
            CrewChange::AddRole(AddRole {
                to: invitation_credential.pub_id(),
                role: invitation_role.to_string(),
            }),
            CrewChange::RevealSecret(RevealSecret {
                secret_kind: CONTENT_SECRET.to_owned(),
                to: invitation_credential.signer().pub_id(),
                encr: invitation_credential
                    .pub_id()
                    .recipient
                    .encrypt_from_anon(&content_secret),
            }),
            CrewChange::RevealSecret(RevealSecret {
                secret_kind: PARTICIPATION_SECRET.to_owned(),
                to: invitation_credential.signer().pub_id(),
                encr: invitation_credential
                    .pub_id()
                    .recipient
                    .encrypt_from_anon(&participation_secret),
            }),
        ])
        .await?;

        Ok(Invitation {
            crew_id: self.id(),
            invitation_role: invitation_role.to_string(),
            invitation_credential,
        })
    }
}

#[derive(Clone, Serialize, Deserialize)]
pub struct Invitation {
    pub crew_id: CrewID,
    pub invitation_role: String,
    pub invitation_credential: MemberCredential,
}