use std::{cell::RefCell, collections::HashSet, rc::Rc};
use audi::Listener;
use credo::{ClaimKind, Credo, GroupSecretRecipientState, GroupState};
use futures::{
channel::{mpsc::channel, oneshot},
future, FutureExt, StreamExt,
};
use jmbl::{ops::OpWithTarget, Input, JMBLViewRef, JMBL};
use mofo::Mofo;
use tlpt::ObjectID;
use crate::{
conventions::{log_key_secret_name, path_for_doc_branch},
managed_jmbl::writing::create_log_as_writable_log,
ScopedDocID,
};
use self::reading::log_stream;
mod reading;
mod writing;
struct ManagedJMBLInner {
content: tlpt::Node,
credo: Credo,
scoped_doc_id: ScopedDocID,
last_secret_state: Option<GroupSecretRecipientState>,
tx_first_secret_state_set: Option<oneshot::Sender<()>>,
logs_already_followed: HashSet<ObjectID>,
state: ManagedJMBLState,
pub jmbl: JMBL,
change_callbacks: Vec<Box<dyn FnMut(bool)>>,
background: Mofo,
}
#[derive(Clone)]
pub struct ManagedJMBL(Rc<RefCell<ManagedJMBLInner>>);
#[derive(Copy, Clone, PartialEq, Eq)]
enum ManagedJMBLState {
Uninitialized,
Loaded,
Writable,
Writing,
}
impl ManagedJMBLInner {
fn receive_op(&mut self, op: OpWithTarget) {
if self.state == ManagedJMBLState::Uninitialized {
self.state = ManagedJMBLState::Loaded;
}
self.jmbl.apply_ops(&Some(op));
for callback in &mut self.change_callbacks {
callback(true);
}
}
}
impl ManagedJMBL {
pub fn id(&self) -> ScopedDocID {
self.0.borrow().scoped_doc_id.clone()
}
pub fn current_root(&self) -> jmbl::Value {
self.0.borrow().jmbl.get_root()
}
pub fn add_change_callback<F>(&mut self, callback: F)
where
F: FnMut(bool) + 'static,
{
(*self.0)
.borrow_mut()
.change_callbacks
.push(Box::new(callback));
}
pub fn load(
scoped_doc_id: ScopedDocID,
content: tlpt::Node,
credo: Credo,
background: Mofo,
) -> (ManagedJMBL, oneshot::Receiver<()>) {
let (team_update_tx, team_update_rx) = channel(100);
let (tx_first_secret_state_set, rx_first_secret_state_set) = oneshot::channel();
background.add_background_task(
Box::pin({
let credo = credo.clone();
let scoped_doc_id = scoped_doc_id.clone();
async move {
credo
.subscribe(
scoped_doc_id.team,
Listener::new(&scoped_doc_id.to_string(), team_update_tx),
)
.await
}
}),
format!("send into team_update_tx {:?}", scoped_doc_id),
);
let managed_jmbl_rc = ManagedJMBL(Rc::new(RefCell::new(ManagedJMBLInner {
content,
credo,
scoped_doc_id: scoped_doc_id.clone(),
last_secret_state: None,
tx_first_secret_state_set: Some(tx_first_secret_state_set),
logs_already_followed: HashSet::new(),
state: ManagedJMBLState::Uninitialized,
jmbl: JMBL::new_empty(),
change_callbacks: Vec::new(),
background: background.clone(),
})));
background.add_background_task(
Box::pin({
let managed_jmbl_rc = managed_jmbl_rc.clone();
team_update_rx.for_each(move |team_update| {
let managed_jmbl_rc = managed_jmbl_rc.clone();
async move {
managed_jmbl_rc.receive_team_update(team_update).await;
}
})
}),
format!("pull out of team_update_tx {:?}", scoped_doc_id),
);
(managed_jmbl_rc, rx_first_secret_state_set)
}
pub async fn create<I: Into<Input>>(
scoped_doc_id: ScopedDocID,
input: I,
content: tlpt::Node,
credo: Credo,
background: Mofo,
) -> ManagedJMBL {
let (managed_jmbl, first_secret_state_set) =
ManagedJMBL::load(scoped_doc_id.clone(), content, credo, background);
first_secret_state_set.await.unwrap();
{
let mut managed_jmbl_ref = (*managed_jmbl.0).borrow_mut();
assert!(managed_jmbl_ref.last_secret_state.is_some());
assert!(managed_jmbl_ref.state == ManagedJMBLState::Uninitialized);
let (writable_log, writing) = create_log_as_writable_log(
managed_jmbl_ref.scoped_doc_id.clone(),
managed_jmbl_ref.content.clone(),
managed_jmbl_ref.credo.clone(),
);
managed_jmbl_ref.background.add_background_task(
writing.boxed_local(),
format!("writing into log {:?}", scoped_doc_id),
);
managed_jmbl_ref.jmbl = JMBL::new_from_root(input, writable_log);
managed_jmbl_ref.state = ManagedJMBLState::Writable;
}
managed_jmbl
}
pub async fn receive_team_update(&self, team_update: GroupState) {
let logs_not_yet_followed = {
let mut self_ref = (*self.0).borrow_mut();
if self_ref.last_secret_state != Some(team_update.secret_state()) {
self_ref.state = match self_ref.state {
ManagedJMBLState::Writing => {
panic!("Tried to receive updates while writing")
}
ManagedJMBLState::Writable => {
self_ref.jmbl.make_readable();
ManagedJMBLState::Loaded
}
current => current,
};
if self_ref.last_secret_state.is_none() {
self_ref
.tx_first_secret_state_set
.take()
.unwrap()
.send(())
.unwrap();
}
self_ref.last_secret_state = Some(team_update.secret_state())
}
let expected_path = path_for_doc_branch(self_ref.scoped_doc_id.clone());
let logs_according_to_update = team_update
.valid_claims
.iter()
.filter_map(|(_id, claim)| {
if let ClaimKind::Statement { path, value } = &claim.expect_v1().kind {
if path == &expected_path {
value.clone().try_into_de::<ObjectID>().ok()
} else {
None
}
} else {
None
}
})
.collect::<HashSet<_>>();
let logs_already_followed = self_ref.logs_already_followed.clone();
let logs_not_yet_followed = logs_according_to_update
.difference(&logs_already_followed)
.cloned()
.collect::<Vec<_>>();
let mut logs_no_longer_valid = logs_already_followed
.difference(&logs_according_to_update)
.peekable();
if logs_no_longer_valid.peek().is_some() {
unimplemented!("Reload document when logs stop being valid");
}
logs_not_yet_followed
};
for log_id in logs_not_yet_followed {
let content = (*self.0).borrow().content.clone();
let scoped_doc_id = (*self.0).borrow().scoped_doc_id.clone();
let (log_tx, log_rx) = channel(100);
content
.add_local_listener(
log_id,
Listener::new(&format!("managed_jmbl_reader_{}", scoped_doc_id), log_tx),
)
.await;
let mut self_ref = (*self.0).borrow_mut();
let log_encr_key = self_ref
.credo
.try_decrypt_entrusted_secret(
&self_ref.scoped_doc_id.team,
&log_key_secret_name(log_id),
)
.expect("Should be able to read log secret once new log is added")
.try_into_de()
.expect("Should be able to deserialize encryption key");
self_ref.background.add_background_task(
Box::pin({
let self_rc = self.clone();
log_stream(log_rx, log_encr_key).for_each(move |op| {
(*self_rc.0).borrow_mut().receive_op(op.unwrap());
future::ready(())
})
}),
format!("reading ops from log {:?}", scoped_doc_id),
);
self_ref.logs_already_followed.insert(log_id);
}
}
pub fn start_writing(&self) -> JMBLViewRef {
let mut self_ref = (*self.0).borrow_mut();
match self_ref.state {
ManagedJMBLState::Writable => {}
ManagedJMBLState::Loaded => {
let (writable_log, writing) = create_log_as_writable_log(
self_ref.scoped_doc_id.clone(),
self_ref.content.clone(),
self_ref.credo.clone(),
);
self_ref.background.add_background_task(
writing.boxed_local(),
format!("writing ops to log {:?}", self_ref.scoped_doc_id),
);
self_ref.jmbl = self_ref.jmbl.switch_to_new_writable_log(writable_log);
self_ref.state = ManagedJMBLState::Writable
}
ManagedJMBLState::Writing => panic!("Already writing"),
ManagedJMBLState::Uninitialized => unreachable!(),
};
self_ref.state = ManagedJMBLState::Writing;
self_ref.jmbl.start_changing_object()
}
pub fn finish_writing(&self, write_view: JMBLViewRef) {
let mut self_ref = (*self.0).borrow_mut();
match self_ref.state {
ManagedJMBLState::Writing => {
self_ref.jmbl.finish_changing_object(write_view);
self_ref.state = ManagedJMBLState::Writable;
for callback in &mut self_ref.change_callbacks {
callback(false);
}
}
_ => panic!("Expected to be writing when finishing writing"),
}
}
}