use std::{
borrow::Borrow,
cell::RefCell,
collections::{HashMap, HashSet},
pin::Pin,
rc::Rc,
};
use crew_rs::{
Crew, CrewChange, CrewID, CrewState, EntrustInfo, MakeStatement, Member, CONTENT_SECRET,
};
use endr::{Diff, ObjectID, SetItem, WriteAccess};
use futures::{
channel::mpsc::UnboundedSender, future, stream, Future, FutureExt, Sink, Stream, StreamExt,
};
use litl::{
impl_debug_as_litl, impl_nested_tagged_data_serde,
raw_data_serde::{raw_data_from_litl, raw_data_to_litl},
NestedTaggedData,
};
use mofo::Mofo;
use objt::objt;
use ridl::{
signing::{Signed, SignerID},
symm_encr::KeySecret,
unauth_symm_encr::UnauthEncryptionStream,
};
use serde_derive::{Deserialize, Serialize};
use tracing::{debug, error, trace, trace_span, warn};
use tracing_futures::Instrument;
use super::Tlpt;
use crate::{
conventions::{log_key_secret_name, CONTENT_TYPE_STR, INCLUDE_LOG_STR, PRESENCE_DOC_STR},
doc_content::{
empty_for_content_type_string, json_stream_content::JsonStreamContent, AnyContentType,
ContentDiff, ContentType,
},
node::WeakTlpt,
team::Team,
TeamID,
};
pub struct DocInner {
tlpt: Tlpt,
pub doc_crew: Crew,
doc_team: TeamID,
endr: endr::Node,
last_member_state: HashSet<Member>,
logs_already_followed: HashSet<ObjectID>,
background: Mofo,
current_write_log: Option<(UnboundedSender<litl::Val>, ObjectID)>,
content_type: AnyContentType,
}
objt!(Doc, WeakDoc, DocInner, doc_crew: Crew, endr: endr::Node);
impl Doc {
fn new(
tlpt: Tlpt,
doc_crew: Crew,
endr: endr::Node,
content_type: AnyContentType,
background: Mofo,
require_intro: bool,
) -> Doc {
let doc_team = doc_crew
.current_state()
.and_then(|state| {
assert_eq!(state.parents.len(), 1);
state.parents.iter().map(|id| TeamID(*id)).next()
})
.expect("Expected doc team to be parent of crew");
let doc = Doc(Rc::new(RefCell::new(DocInner {
tlpt,
doc_crew: doc_crew.clone_ref(),
doc_team,
endr,
last_member_state: HashSet::new(),
logs_already_followed: HashSet::new(),
background: background.clone(),
current_write_log: None,
content_type: content_type.clone(),
})));
content_type.connect_and_init(WeakDoc(Rc::downgrade(&doc.0)), require_intro);
let doc_bg = doc.clone();
background.add_background_task(
doc_crew
.updates("doc".to_owned())
.for_each(move |crew_update| {
let doc = doc_bg.clone();
async move {
if let Some(crew_state) = crew_update.current_state() {
doc.handle_crew_state_update(crew_state).await;
}
}
})
.boxed_local(),
);
doc
}
pub fn tlpt(&self) -> Tlpt {
self.borrow().tlpt.clone()
}
async fn handle_crew_state_update(&self, crew_state: &CrewState) {
let logs_not_yet_followed = {
let mut self_ref = (*self.0).borrow_mut();
let new_member_state = crew_state
.roles
.iter()
.map(|(m, _)| m.clone())
.collect::<HashSet<_>>();
if new_member_state.is_subset(&self_ref.last_member_state) {
self_ref.current_write_log = None;
self_ref.last_member_state = new_member_state
}
let logs_according_to_update = crew_state
.statements
.iter()
.filter_map(|(path, claim, _)| {
if path == INCLUDE_LOG_STR {
litl::from_val::<ObjectID>(claim.clone()).ok()
} 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();
trace!(crew_state = ?self_ref.doc_crew.current_state(), logs_according_to_update = ?logs_according_to_update, logs_already_followed = ?logs_already_followed, logs_not_yet_followed = ?logs_not_yet_followed, "Recevied new set of document logs");
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 endr = (*self.0).borrow().endr.clone();
let doc_crew = (*self.0).borrow().doc_crew.clone();
let diffs = endr.diffs(log_id, format!("doc_reader_{}", doc_crew.id()));
let log_encr_key = litl::from_val(
doc_crew
.get_entrusted_info(CONTENT_SECRET, &log_key_secret_name(log_id))
.expect("Should be able to read log secret once new log is added"),
)
.expect("Should be able to deserialize encryption key");
let mut self_ref = (*self.0).borrow_mut();
let following_task = self_ref.content_type.follow_new_log(
log_id,
decrypting_diff_reader(diffs, log_encr_key).boxed_local(),
);
self_ref.background.add_background_task(following_task);
self_ref.logs_already_followed.insert(log_id);
}
}
pub fn start_writing(&self) -> (UnboundedSender<litl::Val>, endr::ObjectID) {
let tlpt = self.tlpt();
let doc_team = tlpt
.get_loaded_team(&self.doc_team())
.expect("Expected parent team to be loaded when starting to write");
if !tlpt.is_currently_introducted_in(&doc_team) {
let tlpt_for_intro = tlpt.clone();
tlpt.background().add_background_task(
async move { tlpt_for_intro.ensure_introduced_in(&doc_team).await }.boxed_local(),
)
}
self.start_writing_without_intro()
}
pub fn start_writing_without_intro(&self) -> (UnboundedSender<litl::Val>, endr::ObjectID) {
if let Some(current) = &self.borrow().current_write_log {
return current.clone();
}
let (write_log_tx, write_log_rx) = futures::channel::mpsc::unbounded::<litl::Val>();
let endr = self.endr();
let doc_crew = self.doc_crew();
let (log_id, log_write_access, stored_fut) = endr.create_log_sync(Option::<()>::None);
self.borrow()
.background
.add_background_task(Box::pin(stored_fut));
let forward = async move {
let log_encr_key = KeySecret::new_random();
let content_secret = doc_crew
.get_shared_secret(CONTENT_SECRET)
.expect("Should have content secret");
let log_claim_id = doc_crew
.make_changes([
CrewChange::EntrustInfo(EntrustInfo {
to_secret_kind: CONTENT_SECRET.to_owned(),
info_id: log_key_secret_name(log_id),
info: content_secret.encrypt(&log_encr_key).as_encrypted_value(),
}),
CrewChange::MakeStatement(MakeStatement {
path: INCLUDE_LOG_STR.to_owned(),
value: litl::to_val(log_id).unwrap(),
}),
])
.instrument(trace_span!("reveal_log_encr_key", doc_id = ?DocID(doc_crew.id())))
.await
.unwrap();
trace!(secret_claim_ids = ?log_claim_id, "After revealing log secret and adding log");
let encryption_stream = UnauthEncryptionStream::new(log_encr_key, [0; 12].into());
write_log_rx
.scan(
(endr, log_write_access, encryption_stream),
|(endr, log_write_access, encryption_stream), item| {
let mut encrypted = litl::to_vec(&item).unwrap();
encryption_stream.xor_chunk(&mut encrypted);
let entry = litl::Val::string(raw_data_to_litl(&encrypted));
let diff = endr.diff_for_new_log_entry(log_id, &log_write_access, entry);
let endr = endr.clone();
async move {
endr.apply_new_diff(diff).await;
Some(())
}
},
)
.for_each(|_| async {})
.await;
}
.boxed_local();
{
self.borrow().background.add_background_task(forward);
}
self.borrow_mut().current_write_log = Some((write_log_tx.clone(), log_id));
(write_log_tx, log_id)
}
pub fn id(&self) -> DocID {
DocID(self.borrow().doc_crew.id())
}
pub fn content(&self) -> AnyContentType {
self.borrow().content_type.clone()
}
pub fn get_author_of_log(&self, log_id: ObjectID) -> Option<SignerID> {
if let Some(cached) = AUTHOR_OF_LOG_CACHE.with(|cache| cache.borrow().get(&log_id).cloned())
{
return Some(cached);
}
let expected_val = litl::to_val(log_id).unwrap();
self.doc_crew().current_state().and_then(|state| {
state.statements.iter().find_map(|(path, value, made_by)| {
if path == INCLUDE_LOG_STR && value == &expected_val {
AUTHOR_OF_LOG_CACHE.with(|cache| {
cache.borrow_mut().insert(log_id, *made_by);
});
Some(*made_by)
} else {
None
}
})
})
}
pub fn doc_team(&self) -> TeamID {
self.borrow().doc_team
}
pub async fn create_doc_with_same_access(&self, content_type: AnyContentType) -> Doc {
let doc_team_id = self.doc_team();
let tlpt = self.tlpt();
let doc_team = tlpt.load_team(&doc_team_id).await;
tlpt.create_document(&doc_team, content_type).await
}
pub async fn get_presence_doc(&self) -> Option<DocID> {
self.doc_crew().current_state().and_then(|state| {
state.statements.into_iter().find_map(|(path, value, _)| {
if path == PRESENCE_DOC_STR {
Some(litl::from_val::<DocID>(value).unwrap())
} else {
None
}
})
})
}
pub async fn get_or_create_presence_doc(&self) -> Doc {
let existing_doc_id = self.get_presence_doc().await;
let tlpt = self.tlpt();
if let Some(existing_doc_id) = existing_doc_id {
tlpt.load_doc(existing_doc_id).await
} else {
let presence_doc = self
.create_doc_with_same_access(AnyContentType::JsonStream(
JsonStreamContent::new_empty(),
))
.await;
self.doc_crew()
.make_changes([CrewChange::MakeStatement(MakeStatement {
path: PRESENCE_DOC_STR.to_owned(),
value: litl::to_val(presence_doc.id()).unwrap(),
})])
.await
.unwrap();
presence_doc
}
}
pub async fn set_presence(&self, val: litl::Val) {
let presence_doc = self.get_or_create_presence_doc().await;
presence_doc
.content()
.expect_json_stream()
.add_val(val)
.await;
}
pub fn presence_updates(&self) -> impl Stream<Item = (DocID, ObjectID, litl::Val)> {
let tlpt = self.tlpt();
let tlpt_for_future = tlpt.clone();
let self_for_future = self.clone();
let presence_doc = stream::once(async move {
self_for_future
.doc_crew()
.wait_for_state(|state| !state.statements_with_prefix(PRESENCE_DOC_STR).is_empty())
.await;
tlpt_for_future
.load_doc(self_for_future.get_presence_doc().await.unwrap())
.await
});
let updates_per_doc = presence_doc.flat_map(move |presence_doc| {
presence_doc
.content()
.expect_json_stream()
.updates(format!("get-presence-updates-{}", presence_doc.id()))
.map(move |update| (presence_doc.clone(), update))
});
let last_vals_per_presence_log_id = updates_per_doc.flat_map(|(presence_doc, update)| {
stream::iter(
update
.into_iter()
.map(move |(key, val)| (presence_doc.clone(), key, val)),
)
});
let last_vals_per_signer =
last_vals_per_presence_log_id.filter_map(|(presence_doc, key, val)| {
future::ready(if let Some(signer) = presence_doc.get_author_of_log(key) {
Some((presence_doc, signer, key, val))
} else {
warn!("Got presence update for log with unknown signer: {:?}", key);
None
})
});
let tlpt_for_future = tlpt.clone();
let last_vals_per_profile =
last_vals_per_signer.filter_map(move |(presence_doc, signer, key, val)| {
let tlpt_for_future = tlpt_for_future.clone();
async move {
tlpt_for_future
.load_team(&presence_doc.doc_team())
.await
.wait_for_signer_profile(signer)
.await
.map(|author_profile| (author_profile.id(), key, val))
}
});
last_vals_per_profile.boxed_local()
}
}
thread_local! {
pub static AUTHOR_OF_LOG_CACHE: RefCell<HashMap<ObjectID, SignerID>> = RefCell::new(HashMap::new());
}
impl Tlpt {
pub async fn create_document<C: Into<AnyContentType>>(
&self,
parent_team: &Team,
content_type: C,
) -> Doc {
let (doc_team, _) = self.create_team(Some(parent_team.id())).await;
self.ensure_introduced_in(&doc_team).await;
self.create_document_without_intro(&doc_team, content_type)
.await
}
pub async fn create_document_without_intro<C: Into<AnyContentType>>(
&self,
doc_team: &Team,
content_type: C,
) -> Doc {
let endr = self.endr();
let croo = self.croo();
let (doc_crew, _) = croo
.create_crew_with_parents([doc_team.crew.id()], true)
.await;
let content_type = content_type.into();
doc_crew
.make_changes([CrewChange::MakeStatement(MakeStatement {
path: CONTENT_TYPE_STR.to_owned(),
value: litl::Val::str(content_type.content_type()),
})])
.await
.unwrap();
let background = self.background();
let doc = Doc::new(
self.clone(),
doc_crew.clone(),
endr,
content_type,
background,
false,
);
self.insert_doc(doc.clone());
doc
}
pub async fn load_doc(&self, doc_id: DocID) -> Doc {
if let Some(existing) = self.get_doc(&doc_id) {
return existing;
}
let doc_crew = self.croo().load_crew(doc_id.0).await;
let content_type_str = doc_crew
.updates("await_content_type".to_string())
.filter_map(|update| {
future::ready({
update.current_state().and_then(|state| {
state.statements.iter().find_map(|(path, val, _)| {
if path == CONTENT_TYPE_STR {
Some(val.as_str().unwrap().to_owned())
} else {
None
}
})
})
})
})
.next()
.await
.expect("Stream ended before we got content type str");
debug!(
"Got content type str for {}: {:?}",
doc_id, content_type_str
);
let endr = self.endr();
let background = { self.background() };
let content_type = empty_for_content_type_string(&content_type_str)
.unwrap_or_else(|| panic!("Unknown content type: {}", content_type_str));
let doc = Doc::new(
self.clone(),
doc_crew,
endr,
content_type,
background,
false,
);
self.insert_doc(doc.clone());
doc
}
}
#[derive(Copy, Clone, PartialEq, Eq, Hash)]
pub struct DocID(pub CrewID);
impl_nested_tagged_data_serde!(DocID);
impl_debug_as_litl!(DocID);
impl NestedTaggedData for DocID {
const TAG: &'static str = "doc";
type Inner = CrewID;
fn as_inner(&self) -> &Self::Inner {
&self.0
}
fn from_inner(inner: Self::Inner) -> Self
where
Self: Sized,
{
DocID(inner)
}
}
pub fn decrypting_diff_reader(
diffs: impl Stream<Item = Diff>,
log_encr_key: KeySecret,
) -> impl Stream<Item = ContentDiff> {
let mut last_offset = 0;
let mut decryption_stream = UnauthEncryptionStream::new(log_encr_key, [0; 12].into());
diffs
.map(move |diff| match diff {
Diff::Log(log_diff) => {
if log_diff.after < last_offset {
assert!(
log_diff.after + log_diff.new_entries.len() <= last_offset,
"Expected duplicate log diff to fully be in the past"
);
warn!("Duplicate log diff after {}", log_diff.after);
return None;
}
last_offset += log_diff.new_entries.len();
let decrypted_entries = log_diff
.new_entries
.iter()
.map(|entry| {
let mut raw_data = raw_data_from_litl(entry.clone())
.expect("Expected (encrypted) raw data as log entry");
decryption_stream.xor_chunk(&mut raw_data);
litl::from_slice(&raw_data).unwrap()
})
.collect();
Some(ContentDiff {
decrypted_entries,
raw_diff: Diff::Log(log_diff),
})
}
_ => panic!("Expected log diff"),
})
.filter_map(future::ready)
}