diff --git a/Cargo.lock b/Cargo.lock index 678e4f79..b36afa0a 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1740,9 +1740,11 @@ dependencies = [ "nonempty", "olpc-cjson", "once_cell", + "pretty_assertions", "quickcheck", "quickcheck_macros", "radicle-cob", + "radicle-crdt", "radicle-crypto", "radicle-git-ext", "radicle-ssh", @@ -1809,13 +1811,10 @@ name = "radicle-crdt" version = "0.1.0" dependencies = [ "fastrand", - "itertools", "num-traits", "olpc-cjson", - "pretty_assertions", "quickcheck", "quickcheck_macros", - "radicle", "radicle-crypto", "serde", "serde_json", diff --git a/Cargo.toml b/Cargo.toml index 0a9626c3..844202eb 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -13,6 +13,8 @@ members = [ ] default-members = [ "radicle", + "radicle-cob", + "radicle-crdt", "radicle-crypto", "radicle-node", "radicle-ssh", diff --git a/radicle-crdt/Cargo.toml b/radicle-crdt/Cargo.toml index 21ba47ee..a652a8e0 100644 --- a/radicle-crdt/Cargo.toml +++ b/radicle-crdt/Cargo.toml @@ -3,19 +3,24 @@ name = "radicle-crdt" version = "0.1.0" edition = "2021" +[features] +test = ["fastrand"] + [dependencies] +fastrand = { version = "1.8.0", optional = true } num-traits = { version = "0.2.15", default-features = false, features = ["std"] } olpc-cjson = { version = "0.1.1" } -radicle = { path = "../radicle" } serde = { version = "1" } serde_json = { version = "1" } +[dependencies.radicle-crypto] +path = "../radicle-crypto" +version = "0" +features = [] + [dev-dependencies] -itertools = { version = "0.10.5" } fastrand = { version = "1.8.0" } -pretty_assertions = { version = "1.3.0" } quickcheck = { version = "1" } quickcheck_macros = { version = "1" } -radicle = { path = "../radicle", features = ["test"] } radicle-crypto = { path = "../radicle-crypto", features = ["test"] } tempfile = { version = "3" } diff --git a/radicle-crdt/src/change.rs b/radicle-crdt/src/change.rs new file mode 100644 index 00000000..148fe2f9 --- /dev/null +++ b/radicle-crdt/src/change.rs @@ -0,0 +1,130 @@ +use std::collections::BTreeMap; + +use radicle_crypto::{PublicKey, Signature, Signer}; +use serde::{Deserialize, Serialize}; + +use crate::clock::LClock; + +/// Identifies a change. +pub type ChangeId = (LClock, Author); +/// The author of a change. +pub type Author = PublicKey; +/// Alias for `Author`. +pub type ActorId = PublicKey; + +/// The `Change` is the unit of replication. +/// Everything that can be done in the system is represented by a `Change` object. +/// Changes are applied to an accumulator to yield a final state. +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +pub struct Change { + /// The action carried out by this change. + pub action: A, + /// The author of the change. + pub author: Author, + /// Lamport clock. + pub clock: LClock, +} + +impl<'de, A: Serialize + Deserialize<'de>> Change { + /// Get the change id. + pub fn id(&self) -> ChangeId { + (self.clock, self.author) + } + + /// Serialize the change into a byte string. + pub fn encode(&self) -> Vec { + let mut buf = Vec::new(); + let mut serializer = + serde_json::Serializer::with_formatter(&mut buf, olpc_cjson::CanonicalFormatter::new()); + + self.serialize(&mut serializer).unwrap(); + + buf + } + + /// Deserialize a change from a byte string. + pub fn decode(bytes: &'de [u8]) -> Result { + serde_json::from_slice(bytes) + } +} + +/// Change envelope. Carries signed changes. +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +pub struct Envelope { + /// Changes included in this envelope, serialized as JSON. + pub changes: Vec, + /// Signature over the change, by the change author. + pub signature: Signature, +} + +/// An object that can be used to create and sign changes. +#[derive(Default)] +pub struct Actor { + pub signer: G, + pub clock: LClock, + pub changes: BTreeMap<(LClock, PublicKey), Change>, +} + +impl Actor { + pub fn new(signer: G) -> Self { + Self { + signer, + clock: LClock::default(), + changes: BTreeMap::default(), + } + } + + pub fn receive(&mut self, changes: impl IntoIterator>) -> LClock { + for change in changes { + let clock = change.clock; + + self.changes.insert((clock, change.author), change); + self.clock.merge(clock); + } + self.clock + } + + /// Reset actor state to initial state. + pub fn reset(&mut self) { + self.changes.clear(); + self.clock = LClock::default(); + } + + /// Returned an ordered list of events. + pub fn timeline(&self) -> impl Iterator> { + self.changes.values() + } + + /// Create a new change. + pub fn change(&mut self, action: A) -> Change { + let author = *self.signer.public_key(); + let clock = self.clock.tick(); + let change = Change { + action, + author, + clock, + }; + self.changes.insert((self.clock, author), change.clone()); + + change + } + + pub fn sign(&self, changes: impl IntoIterator>) -> Envelope { + let changes = changes.into_iter().collect::>(); + let json = serde_json::to_value(changes).unwrap(); + + let mut buffer = Vec::new(); + let mut serializer = serde_json::Serializer::with_formatter( + &mut buffer, + olpc_cjson::CanonicalFormatter::new(), + ); + json.serialize(&mut serializer).unwrap(); + + let signature = self.signer.sign(&buffer); + + Envelope { + changes: buffer, + signature, + } + } +} diff --git a/radicle-crdt/src/lib.rs b/radicle-crdt/src/lib.rs index 6ca7bc54..bd09ea88 100644 --- a/radicle-crdt/src/lib.rs +++ b/radicle-crdt/src/lib.rs @@ -1,16 +1,20 @@ #![allow(clippy::collapsible_if)] #![allow(clippy::collapsible_else_if)] #![allow(clippy::type_complexity)] +pub mod change; pub mod clock; pub mod lwwmap; pub mod lwwreg; pub mod lwwset; pub mod ord; pub mod redactable; -pub mod thread; -#[cfg(test)] -mod test; +#[cfg(any(test, feature = "test"))] +pub mod test; + +//////////////////////////////////////////////////////////////////////////////// + +pub use change::*; //////////////////////////////////////////////////////////////////////////////// diff --git a/radicle-crdt/src/test.rs b/radicle-crdt/src/test.rs index 9b51ce17..07018ef0 100644 --- a/radicle-crdt/src/test.rs +++ b/radicle-crdt/src/test.rs @@ -55,7 +55,10 @@ impl<'a, T, C: Default> WeightedGenerator<'a, T, C> { pub fn assert_laws(a: &S, b: &S, c: &S) { assert_associative(a, b, c); assert_commutative(a, b); + assert_commutative(b, c); assert_idempotent(a); + assert_idempotent(b); + assert_idempotent(c); } pub fn assert_associative(a: &S, b: &S, c: &S) { diff --git a/radicle-crdt/src/thread.rs b/radicle-crdt/src/thread.rs deleted file mode 100644 index f18fac95..00000000 --- a/radicle-crdt/src/thread.rs +++ /dev/null @@ -1,645 +0,0 @@ -use std::collections::BTreeMap; -use std::ops::Deref; - -use serde::{Deserialize, Serialize}; - -use radicle::cob::common::Reaction; -use radicle::cob::Timestamp; -use radicle::crypto::{PublicKey, Signature, Signer}; -use radicle::hash; - -use crate::clock::LClock; -use crate::lwwreg::LWWReg; -use crate::lwwset::LWWSet; -use crate::redactable::Redactable; -use crate::Semilattice; - -/// Identifies a change. -pub type ChangeId = radicle::hash::Digest; -/// Identifies a tag. -pub type TagId = String; -/// The author of a change. -pub type Author = PublicKey; -/// Alias for `Author`. -pub type ActorId = PublicKey; - -/// The `Change` is the unit of replication. -/// Everything that can be done in the system is represented by a `Change` object. -/// Changes are applied to an accumulator to yield a final state. -#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] -pub struct Change { - /// The action carried out by this change. - action: Action, - /// The author of the change. - author: Author, - /// The time at which this change was authored. - timestamp: Timestamp, - /// Lamport clock. - clock: LClock, -} - -impl Change { - /// Get the change id. - pub fn id(&self) -> ChangeId { - hash::Digest::new(self.encode()) - } - - /// Serialize the change into a byte string. - pub fn encode(&self) -> Vec { - let mut buf = Vec::new(); - let mut serializer = - serde_json::Serializer::with_formatter(&mut buf, olpc_cjson::CanonicalFormatter::new()); - - self.serialize(&mut serializer).unwrap(); - - buf - } - - /// Deserialize a change from a byte string. - pub fn decode(bytes: &[u8]) -> Result { - serde_json::from_slice(bytes) - } -} - -/// Change envelope. Carries signed changes. -#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] -pub struct Envelope { - /// Changes included in this envelope, serialized as JSON. - pub changes: Vec, - /// Signature over the change, by the change author. - pub signature: Signature, -} - -/// A comment on a discussion thread. -#[derive(Default, Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] -pub struct Comment { - /// The comment body. - body: String, - /// Thread or comment this is a reply to. - reply_to: Option, -} - -impl Comment { - /// Create a new comment. - pub fn new(body: String, reply_to: Option) -> Self { - Self { body, reply_to } - } -} - -impl PartialOrd for Comment { - fn partial_cmp(&self, _other: &Self) -> Option { - None - } -} - -/// An action that can be carried out in a change. -#[derive(Debug, PartialEq, Eq, Clone, Serialize, Deserialize)] -pub enum Action { - /// Comment on a thread. - Comment { comment: Comment }, - /// Redact a change. Not all changes can be redacted. - Redact { id: ChangeId }, - /// Add a tag to the thread. - Tag { tag: TagId }, - /// Remove a tag from the thread. - Untag { tag: TagId }, - /// React to a change. - React { - to: ChangeId, - reaction: Reaction, - active: bool, - }, -} - -/// A discussion thread. -#[derive(Debug, Default, Clone, PartialEq, Eq)] -pub struct Thread { - /// The comments under the thread. - comments: BTreeMap>, - /// Associated tags. - tags: BTreeMap>, - /// Reactions to changes. - reactions: BTreeMap>, -} - -impl Semilattice for Thread { - fn merge(&mut self, other: Self) { - self.comments.merge(other.comments); - self.tags.merge(other.tags); - self.reactions.merge(other.reactions); - } -} - -impl Deref for Thread { - type Target = BTreeMap>; - - fn deref(&self) -> &Self::Target { - &self.comments - } -} - -impl Thread { - pub fn clear(&mut self) { - self.comments.clear(); - } - - pub fn apply(&mut self, changes: impl IntoIterator) { - for change in changes.into_iter() { - let id = change.id(); - - match change.action { - Action::Comment { comment } => { - match self.comments.get(&id) { - Some(Redactable::Present(_)) => { - // Do nothing, the action was already processed, - // since a change with the same content-id as this - // one exists already. - } - Some(Redactable::Redacted) => { - // Do nothing, the action was redacted. - } - None => { - self.comments.insert(id, Redactable::Present(comment)); - } - } - } - Action::Redact { id } => { - self.comments.insert(id, Redactable::Redacted); - } - Action::Tag { tag } => { - self.tags - .entry(tag) - .and_modify(|r| r.set(true, change.timestamp)) - .or_insert_with(|| LWWReg::new(true, change.timestamp)); - } - Action::Untag { tag } => { - self.tags - .entry(tag) - .and_modify(|r| r.set(false, change.timestamp)) - .or_insert_with(|| LWWReg::new(false, change.timestamp)); - } - Action::React { - to, - reaction, - active, - } => { - self.reactions - .entry(to) - .and_modify(|reactions| { - if active { - reactions.insert((change.author, reaction), change.timestamp); - } else { - reactions.remove((change.author, reaction), change.timestamp); - } - }) - .or_insert_with(|| { - if active { - LWWSet::singleton((change.author, reaction), change.timestamp) - } else { - let mut set = LWWSet::default(); - set.remove((change.author, reaction), change.timestamp); - set - } - }); - } - } - } - } - - pub fn comments(&self) -> impl Iterator + '_ { - self.comments.iter().filter_map(|(id, comment)| { - if let Redactable::Present(c) = comment { - Some((id, c)) - } else { - None - } - }) - } - - pub fn tags(&self) -> impl Iterator + '_ { - self.tags - .iter() - .filter_map(|(tag, r)| if *r.get() { Some(tag) } else { None }) - } -} - -/// An object that can be used to create and sign changes. -#[derive(Default)] -pub struct Actor { - signer: G, - clock: LClock, - changes: BTreeMap<(LClock, PublicKey), Change>, -} - -impl Actor { - pub fn new(signer: G) -> Self { - Self { - signer, - clock: LClock::default(), - changes: BTreeMap::default(), - } - } - - pub fn receive(&mut self, changes: impl IntoIterator) -> LClock { - for change in changes { - let clock = change.clock; - - self.changes.insert((clock, change.author), change); - self.clock.merge(clock); - } - self.clock - } - - /// Reset actor state to initial state. - pub fn reset(&mut self) { - self.changes.clear(); - self.clock = LClock::default(); - } - - /// Create a new thread. - pub fn thread(&self) -> Thread { - Thread::default() - } - - /// Returned an ordered list of events. - pub fn timeline(&self) -> impl Iterator { - self.changes.values() - } - - /// Create a new comment. - pub fn comment( - &mut self, - body: &str, - timestamp: Timestamp, - parent: Option, - ) -> Change { - self.change( - Action::Comment { - comment: Comment::new(String::from(body), parent), - }, - timestamp, - ) - } - - /// Add a tag. - pub fn tag(&mut self, tag: TagId, timestamp: Timestamp) -> Change { - self.change(Action::Tag { tag }, timestamp) - } - - /// Remove a tag. - pub fn untag(&mut self, tag: TagId, timestamp: Timestamp) -> Change { - self.change(Action::Untag { tag }, timestamp) - } - - /// Create a new redaction. - pub fn redact(&mut self, id: ChangeId, timestamp: Timestamp) -> Change { - self.change(Action::Redact { id }, timestamp) - } - - /// Create a new change. - pub fn change(&mut self, action: Action, timestamp: Timestamp) -> Change { - let author = *self.signer.public_key(); - let clock = self.clock.tick(); - let change = Change { - action, - author, - timestamp, - clock, - }; - self.changes.insert((self.clock, author), change.clone()); - - change - } - - pub fn sign(&self, changes: impl IntoIterator) -> Envelope { - let changes = changes.into_iter().collect::>(); - let json = serde_json::to_value(changes).unwrap(); - - let mut buffer = Vec::new(); - let mut serializer = serde_json::Serializer::with_formatter( - &mut buffer, - olpc_cjson::CanonicalFormatter::new(), - ); - json.serialize(&mut serializer).unwrap(); - - let signature = self.signer.sign(&buffer); - - Envelope { - changes: buffer, - signature, - } - } -} - -#[cfg(test)] -mod tests { - use std::ops::ControlFlow; - use std::str::FromStr; - use std::{array, iter}; - - use itertools::Itertools; - use pretty_assertions::assert_eq; - use quickcheck::Arbitrary; - use quickcheck_macros::quickcheck; - use radicle::{cob::TypeName, crypto::test::signer::MockSigner, identity::project::Identity}; - - use super::*; - use crate::test::{assert_laws, WeightedGenerator}; - - #[derive(Clone)] - struct Changes { - permutations: [Vec; N], - } - - impl std::fmt::Debug for Changes { - fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { - for (i, p) in self.permutations.iter().enumerate() { - writeln!( - f, - "{i}: {:#?}", - p.iter().map(|c| &c.action).collect::>() - )?; - } - Ok(()) - } - } - - impl Arbitrary for Changes { - fn arbitrary(g: &mut quickcheck::Gen) -> Self { - let rng = fastrand::Rng::with_seed(u64::arbitrary(g)); - let gen = WeightedGenerator::, Vec)>::new(rng.clone()) - .variant(2, |_, rng| { - Some(Action::Comment { - comment: Comment { - body: iter::repeat_with(|| rng.alphabetic()).take(16).collect(), - reply_to: Default::default(), - }, - }) - }) - .variant(2, |(_, changes), rng| { - if changes.is_empty() { - return None; - } - let to = changes[rng.usize(..changes.len())].id(); - - Some(Action::React { - to, - reaction: Reaction::new('✨').unwrap(), - active: rng.bool(), - }) - }) - .variant(2, |(_, changes), rng| { - if changes.is_empty() { - return None; - } - let id = changes[rng.usize(..changes.len())].id(); - Some(Action::Redact { id }) - }) - .variant(2, |(tags, _), rng| { - let tag = if tags.is_empty() || rng.bool() { - let tag = iter::repeat_with(|| rng.alphabetic()) - .take(8) - .collect::(); - tags.push(tag.clone()); - tag - } else { - tags[rng.usize(..tags.len())].clone() - }; - Some(Action::Tag { tag }) - }) - .variant(2, |(tags, _), rng| { - if tags.is_empty() { - return None; - } - let tag = tags[rng.usize(..tags.len())].clone(); - Some(Action::Untag { tag }) - }); - - let mut changes = Vec::new(); - let mut permutations: [Vec; N] = array::from_fn(|_| Vec::new()); - let mut clock = LClock::default(); - let author = PublicKey::from([0; 32]); - - for action in gen.take(g.size().min(8)) { - let timestamp = Timestamp::now() + rng.u64(0..3); - let clock = clock.tick(); - - changes.push(Change { - action, - author, - timestamp, - clock, - }); - } - - for p in &mut permutations { - *p = changes.clone(); - rng.shuffle(&mut changes); - } - - Changes { permutations } - } - } - - #[test] - fn test_storage() { - let tmp = tempfile::tempdir().unwrap(); - let (_storage, signer, repository) = radicle::test::setup::context(&tmp); - let mut alice = Actor::new(signer); - let project = Identity::load(alice.signer.public_key(), &repository).unwrap(); - let timestamp = Timestamp::now(); - let typename = TypeName::from_str("xyz.radicle.thread").unwrap(); - - let a1 = alice.comment("First comment", timestamp + 1, None); - let a2 = alice.comment("Second comment", timestamp + 2, None); - - let mut expected = Thread::default(); - expected.apply([a1.clone(), a2.clone()]); - - let created = radicle::cob::create( - &repository, - &alice.signer, - &project, - radicle::cob::Create { - author: None, - history_type: radicle::cob::HistoryType::default(), - contents: a1.encode(), - typename: typename.clone(), - message: "Thread created".to_owned(), - }, - ) - .unwrap(); - - radicle::cob::update( - &repository, - &alice.signer, - &project, - radicle::cob::Update { - author: None, - history_type: radicle::cob::HistoryType::default(), - changes: a2.encode(), - object_id: *created.id(), - typename: typename.clone(), - message: "Thread updated".to_owned(), - }, - ) - .unwrap(); - - let retrieved = radicle::cob::get(&repository, &typename, created.id()) - .unwrap() - .unwrap(); - - let actual: Thread = retrieved - .history() - .traverse(Thread::default(), |mut acc, entry| { - let change = Change::decode(entry.contents()).unwrap(); - acc.apply([change]); - - ControlFlow::Continue(acc) - }); - - assert_eq!(actual, expected); - } - - #[test] - fn test_timelines_basic() { - let mut alice = Actor::::default(); - let mut bob = Actor::::default(); - let timestamp = Timestamp::now(); - - let a1 = alice.comment("First comment", timestamp + 1, None); - let a2 = alice.comment("Second comment", timestamp + 2, None); - - bob.receive([a1.clone(), a2.clone()]); - assert_eq!( - bob.timeline().collect::>(), - alice.timeline().collect::>() - ); - assert_eq!(alice.timeline().collect::>(), vec![&a1, &a2]); - - bob.reset(); - bob.receive([a2, a1]); - assert_eq!( - bob.timeline().collect::>(), - alice.timeline().collect::>() - ); - } - - #[test] - fn test_timelines_concurrent() { - let mut alice = Actor::::default(); - let mut bob = Actor::::default(); - let mut eve = Actor::::default(); - let timestamp = Timestamp::now(); - - let a1 = alice.comment("First comment", timestamp, None); - - bob.receive([a1.clone()]); - - let b0 = bob.comment("Bob's first reply to Alice", timestamp, None); - let b1 = bob.comment("Bob's second reply to Alice", timestamp, None); - - eve.receive([b1.clone(), b0.clone()]); - let e0 = eve.comment("Eve's first reply to Alice", timestamp, None); - - bob.receive([e0.clone()]); - let b2 = bob.comment("Bob's third reply to Alice", timestamp, None); - - eve.receive([b2.clone(), a1.clone()]); - let e1 = eve.comment("Eve's second reply to Alice", timestamp, None); - - alice.receive([b0.clone(), b1.clone(), b2.clone(), e0.clone(), e1.clone()]); - bob.receive([e1.clone()]); - - let a2 = alice.comment("Second comment", timestamp, None); - eve.receive([a2.clone()]); - bob.receive([a2.clone()]); - - assert_eq!(alice.changes.len(), 7); - assert_eq!(bob.changes.len(), 7); - assert_eq!(eve.changes.len(), 7); - - assert_eq!( - bob.timeline().collect::>(), - alice.timeline().collect::>() - ); - assert_eq!( - eve.timeline().collect::>(), - alice.timeline().collect::>() - ); - assert_eq!( - vec![&a1, &b0, &b1, &e0, &b2, &e1, &a2], - alice.timeline().collect::>(), - ); - } - - #[quickcheck] - fn prop_invariants(log: Changes<3>) { - let t = Thread::default(); - let [p1, p2, p3] = log.permutations; - - let mut t1 = t.clone(); - t1.apply(p1); - - let mut t2 = t.clone(); - t2.apply(p2); - - let mut t3 = t; - t3.apply(p3); - - assert_eq!(t1, t2); - assert_eq!(t2, t3); - assert_laws(&t1, &t2, &t3); - } - - #[test] - fn test_invariants() { - let mut alice = Actor::::default(); - let bob = Actor::::default(); - let time = Timestamp::now(); - - let t = bob.thread(); - let a0 = alice.comment("Ham", time, None); - let a1 = alice.comment("Rye", time, None); - let a2 = alice.comment("Dough", time, Some(a1.id())); - let a3 = alice.redact(a1.id(), time); - let a4 = alice.comment("Bread", time, None); - - assert_order_invariance(&t, [&a0, &a1, &a2, &a3, &a4]); - assert_idempotence(&t, [&a0, &a1, &a2, &a3, &a4]); - } - - fn assert_order_invariance<'a>(t: &Thread, changes: impl IntoIterator) { - let changes = changes.into_iter().cloned().collect::>(); - let count = changes.len(); - - let mut actual = t.clone(); - let mut expected = t.clone(); - expected.clear(); - expected.apply(changes.clone()); - - for permutation in changes.into_iter().permutations(count) { - actual.clear(); - actual.apply(permutation); - - assert_eq!(actual, expected); - } - } - - fn assert_idempotence<'a>(t: &Thread, changes: impl IntoIterator) { - let changes = changes.into_iter().cloned().collect::>(); - - let mut actual = t.clone(); - let mut expected = t.clone(); - - expected.clear(); - expected.apply(changes.clone()); - - actual.clear(); - actual.apply(changes.clone()); - actual.apply(changes.clone()); - actual.apply(changes); - - assert_eq!(actual, expected); - } -} diff --git a/radicle/Cargo.toml b/radicle/Cargo.toml index eaa3fbd1..4a90108c 100644 --- a/radicle/Cargo.toml +++ b/radicle/Cargo.toml @@ -44,6 +44,10 @@ features = ["vendored-libgit2"] path = "../radicle-cob" version = "0" +[dependencies.radicle-crdt] +path = "../radicle-crdt" +version = "0" + [dependencies.radicle-crypto] path = "../radicle-crypto" version = "0" @@ -60,6 +64,7 @@ default-features = false optional = true [dev-dependencies] +pretty_assertions = { version = "1.3.0" } quickcheck_macros = { version = "1", default-features = false } quickcheck = { version = "1", default-features = false } @@ -67,3 +72,8 @@ quickcheck = { version = "1", default-features = false } path = "../radicle-crypto" version = "0" features = ["test"] + +[dev-dependencies.radicle-crdt] +path = "../radicle-crdt" +version = "0" +features = ["test"] diff --git a/radicle/src/cob.rs b/radicle/src/cob.rs index fde7ed9c..96683bbc 100644 --- a/radicle/src/cob.rs +++ b/radicle/src/cob.rs @@ -2,6 +2,8 @@ pub mod automerge; pub mod common; pub mod issue; pub mod patch; +pub mod store; +pub mod thread; pub use cob::{ identity, object::collaboration::error, CollaborativeObject, Contents, Create, Entry, History, diff --git a/radicle/src/cob/store.rs b/radicle/src/cob/store.rs new file mode 100644 index 00000000..f88a8114 --- /dev/null +++ b/radicle/src/cob/store.rs @@ -0,0 +1,149 @@ +//! Generic COB storage. +#![allow(clippy::large_enum_variant)] +use std::marker::PhantomData; + +use crate::cob; +use crate::cob::common::Author; +use crate::cob::CollaborativeObject; +use crate::cob::{Contents, Create, History, HistoryType, ObjectId, TypeName, Update}; +use crate::crypto::PublicKey; +use crate::git; +use crate::identity::project; +use crate::prelude::*; +use crate::storage::git as storage; + +/// A type that can be materialized from an event history. +/// All collaborative objects implement this trait. +pub trait FromHistory: Sized { + /// The object type name. + fn type_name() -> &'static TypeName; + /// Create an object from a history. + fn from_history(history: &History) -> Result; +} + +/// Store error. +#[derive(Debug, thiserror::Error)] +pub enum Error { + #[error("create error: {0}")] + Create(#[from] cob::error::Create), + #[error("update error: {0}")] + Update(#[from] cob::error::Update), + #[error("retrieve error: {0}")] + Retrieve(#[from] cob::error::Retrieve), + #[error(transparent)] + Identity(#[from] project::IdentityError), + #[error("object `{1}`of type `{0}` was not found")] + NotFound(TypeName, ObjectId), +} + +/// Storage for collaborative objects of a specific type `T` in a single project. +pub struct Store<'a, T> { + whoami: PublicKey, + project: project::Identity, + raw: &'a storage::Repository, + witness: PhantomData, +} + +impl<'a, T> AsRef for Store<'a, T> { + fn as_ref(&self) -> &storage::Repository { + self.raw + } +} + +impl<'a, T> Store<'a, T> { + /// Open a new generic store. + pub fn open(whoami: PublicKey, store: &'a storage::Repository) -> Result { + let project = project::Identity::load(&whoami, store)?; + + Ok(Self { + project, + whoami, + raw: store, + witness: PhantomData, + }) + } +} + +impl<'a, T> Store<'a, T> { + /// Get this store's author. + pub fn author(&self) -> Author { + Author::new(self.whoami) + } + + /// Get the public key associated with this store. + pub fn public_key(&self) -> &PublicKey { + &self.whoami + } +} + +impl<'a, T: FromHistory> Store<'a, T> { + /// Update an object. + pub fn update( + &self, + object_id: ObjectId, + message: &'static str, + changes: Contents, + signer: &G, + ) -> Result { + cob::update( + self.raw, + signer, + &self.project, + Update { + author: Some(cob::Author::from(*signer.public_key())), + object_id, + history_type: HistoryType::default(), + typename: T::type_name().clone(), + message: message.to_owned(), + changes, + }, + ) + } + + /// Create an object. + pub fn create( + &self, + message: &'static str, + contents: Contents, + signer: &G, + ) -> Result { + cob::create( + self.raw, + signer, + &self.project, + Create { + author: Some(cob::Author::from(*signer.public_key())), + history_type: HistoryType::default(), + typename: T::type_name().clone(), + message: message.to_owned(), + contents, + }, + ) + } + + /// Get an object. + pub fn get(&self, id: &ObjectId) -> Result, Error> { + let cob = cob::get(self.raw, T::type_name(), id)?; + + if let Some(cob) = cob { + let history = cob.history(); + let obj = T::from_history(history)?; + + Ok(Some(obj)) + } else { + Ok(None) + } + } + + /// List objects. + pub fn list(&self) -> Result, Error> { + let raw = cob::list(self.raw, T::type_name())?; + + raw.into_iter() + .map(|o| { + let obj = T::from_history(o.history())?; + Ok::<_, Error>((*o.id(), obj)) + }) + .collect() + } +} diff --git a/radicle/src/cob/thread.rs b/radicle/src/cob/thread.rs new file mode 100644 index 00000000..b884b247 --- /dev/null +++ b/radicle/src/cob/thread.rs @@ -0,0 +1,482 @@ +use std::collections::BTreeMap; +use std::ops::{ControlFlow, Deref, DerefMut}; +use std::str::FromStr; + +use once_cell::sync::Lazy; +use radicle_crdt as crdt; +use serde::{Deserialize, Serialize}; + +use crate::cob::common::Reaction; +use crate::cob::store; +use crate::cob::{History, TypeName}; +use crate::crypto::{PublicKey, Signer}; + +use crdt::clock::LClock; +use crdt::lwwreg::LWWReg; +use crdt::lwwset::LWWSet; +use crdt::redactable::Redactable; +use crdt::{Change, ChangeId, Semilattice}; + +/// Type name of a thread. +pub static TYPENAME: Lazy = + Lazy::new(|| FromStr::from_str("xyz.radicle.thread").expect("type name is valid")); + +/// Identifies a tag. +pub type TagId = String; +/// Alias for `Author`. +pub type ActorId = PublicKey; + +/// A comment on a discussion thread. +#[derive(Default, Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +pub struct Comment { + /// The comment body. + body: String, + /// Thread or comment this is a reply to. + reply_to: Option, +} + +impl Comment { + /// Create a new comment. + pub fn new(body: String, reply_to: Option) -> Self { + Self { body, reply_to } + } +} + +impl PartialOrd for Comment { + fn partial_cmp(&self, _other: &Self) -> Option { + None + } +} + +/// An action that can be carried out in a change. +#[derive(Debug, PartialEq, Eq, Clone, Serialize, Deserialize)] +pub enum Action { + /// Comment on a thread. + Comment { comment: Comment }, + /// Redact a change. Not all changes can be redacted. + Redact { id: ChangeId }, + /// Add a tag to the thread. + Tag { tag: TagId }, + /// Remove a tag from the thread. + Untag { tag: TagId }, + /// React to a change. + React { + to: ChangeId, + reaction: Reaction, + active: bool, + }, +} + +/// A discussion thread. +#[derive(Debug, Default, Clone, PartialEq, Eq)] +pub struct Thread { + /// The comments under the thread. + comments: BTreeMap>, + /// Associated tags. + tags: BTreeMap>, + /// Reactions to changes. + reactions: BTreeMap>, +} + +impl store::FromHistory for Thread { + fn type_name() -> &'static radicle_cob::TypeName { + &*TYPENAME + } + + fn from_history(history: &History) -> Result { + Ok(history.traverse(Thread::default(), |mut acc, entry| { + if let Ok(change) = Change::decode(entry.contents()) { + acc.apply([change]); + ControlFlow::Continue(acc) + } else { + ControlFlow::Break(acc) + } + })) + } +} +impl Semilattice for Thread { + fn merge(&mut self, other: Self) { + self.comments.merge(other.comments); + self.tags.merge(other.tags); + self.reactions.merge(other.reactions); + } +} + +impl Deref for Thread { + type Target = BTreeMap>; + + fn deref(&self) -> &Self::Target { + &self.comments + } +} + +impl Thread { + pub fn clear(&mut self) { + self.comments.clear(); + } + + pub fn apply(&mut self, changes: impl IntoIterator>) { + for change in changes.into_iter() { + let id = change.id(); + + match change.action { + Action::Comment { comment } => { + match self.comments.get(&id) { + Some(Redactable::Present(_)) => { + // Do nothing, the action was already processed, + // since a change with the same content-id as this + // one exists already. + } + Some(Redactable::Redacted) => { + // Do nothing, the action was redacted. + } + None => { + self.comments.insert(id, Redactable::Present(comment)); + } + } + } + Action::Redact { id } => { + self.comments.insert(id, Redactable::Redacted); + } + Action::Tag { tag } => { + self.tags + .entry(tag) + .and_modify(|r| r.set(true, change.clock)) + .or_insert_with(|| LWWReg::new(true, change.clock)); + } + Action::Untag { tag } => { + self.tags + .entry(tag) + .and_modify(|r| r.set(false, change.clock)) + .or_insert_with(|| LWWReg::new(false, change.clock)); + } + Action::React { + to, + reaction, + active, + } => { + self.reactions + .entry(to) + .and_modify(|reactions| { + if active { + reactions.insert((change.author, reaction), change.clock); + } else { + reactions.remove((change.author, reaction), change.clock); + } + }) + .or_insert_with(|| { + if active { + LWWSet::singleton((change.author, reaction), change.clock) + } else { + let mut set = LWWSet::default(); + set.remove((change.author, reaction), change.clock); + set + } + }); + } + } + } + } + + pub fn comments(&self) -> impl Iterator + '_ { + self.comments.iter().filter_map(|(id, comment)| { + if let Redactable::Present(c) = comment { + Some((id, c)) + } else { + None + } + }) + } + + pub fn tags(&self) -> impl Iterator + '_ { + self.tags + .iter() + .filter_map(|(tag, r)| if *r.get() { Some(tag) } else { None }) + } +} + +/// An object that can be used to create and sign changes. +pub struct Actor { + inner: crdt::Actor, +} + +impl Default for Actor { + fn default() -> Self { + Self { + inner: crdt::Actor::new(G::default()), + } + } +} + +impl Actor { + pub fn new(signer: G) -> Self { + Self { + inner: crdt::Actor::new(signer), + } + } + + /// Create a new thread. + pub fn thread(&self) -> Thread { + Thread::default() + } + + /// Create a new comment. + pub fn comment(&mut self, body: &str, parent: Option) -> Change { + self.change(Action::Comment { + comment: Comment::new(String::from(body), parent), + }) + } + + /// Add a tag. + pub fn tag(&mut self, tag: TagId) -> Change { + self.change(Action::Tag { tag }) + } + + /// Remove a tag. + pub fn untag(&mut self, tag: TagId) -> Change { + self.change(Action::Untag { tag }) + } + + /// Create a new redaction. + pub fn redact(&mut self, id: ChangeId) -> Change { + self.change(Action::Redact { id }) + } +} + +impl Deref for Actor { + type Target = crdt::Actor; + + fn deref(&self) -> &Self::Target { + &self.inner + } +} + +impl DerefMut for Actor { + fn deref_mut(&mut self) -> &mut Self::Target { + &mut self.inner + } +} + +#[cfg(test)] +mod tests { + use std::{array, iter}; + + use crate::crypto::test::signer::MockSigner; + use pretty_assertions::assert_eq; + use quickcheck::Arbitrary; + use quickcheck_macros::quickcheck; + + use super::*; + use crate as radicle; + use crdt::test::{assert_laws, WeightedGenerator}; + + #[derive(Clone)] + struct Changes { + permutations: [Vec>; N], + } + + impl std::fmt::Debug for Changes { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + for (i, p) in self.permutations.iter().enumerate() { + writeln!( + f, + "{i}: {:#?}", + p.iter().map(|c| &c.action).collect::>() + )?; + } + Ok(()) + } + } + + impl Arbitrary for Changes { + fn arbitrary(g: &mut quickcheck::Gen) -> Self { + let rng = fastrand::Rng::with_seed(u64::arbitrary(g)); + let gen = + WeightedGenerator::, Vec>)>::new(rng.clone()) + .variant(2, |_, rng| { + Some(Action::Comment { + comment: Comment { + body: iter::repeat_with(|| rng.alphabetic()).take(16).collect(), + reply_to: Default::default(), + }, + }) + }) + .variant(2, |(_, changes), rng| { + if changes.is_empty() { + return None; + } + let to = changes[rng.usize(..changes.len())].id(); + + Some(Action::React { + to, + reaction: Reaction::new('✨').unwrap(), + active: rng.bool(), + }) + }) + .variant(2, |(_, changes), rng| { + if changes.is_empty() { + return None; + } + let id = changes[rng.usize(..changes.len())].id(); + Some(Action::Redact { id }) + }) + .variant(2, |(tags, _), rng| { + let tag = if tags.is_empty() || rng.bool() { + let tag = iter::repeat_with(|| rng.alphabetic()) + .take(8) + .collect::(); + tags.push(tag.clone()); + tag + } else { + tags[rng.usize(..tags.len())].clone() + }; + Some(Action::Tag { tag }) + }) + .variant(2, |(tags, _), rng| { + if tags.is_empty() { + return None; + } + let tag = tags[rng.usize(..tags.len())].clone(); + Some(Action::Untag { tag }) + }); + + let mut changes = Vec::new(); + let mut permutations: [Vec>; N] = array::from_fn(|_| Vec::new()); + let mut clock = LClock::default(); + let author = PublicKey::from([0; 32]); + + for action in gen.take(g.size().min(8)) { + let clock = clock.tick(); + + changes.push(Change { + action, + author, + clock, + }); + } + + for p in &mut permutations { + *p = changes.clone(); + rng.shuffle(&mut changes); + } + + Changes { permutations } + } + } + + #[test] + fn test_storage() { + let tmp = tempfile::tempdir().unwrap(); + let (_, signer, repository) = radicle::test::setup::context(&tmp); + let store = + radicle::cob::store::Store::::open(*signer.public_key(), &repository).unwrap(); + + let mut alice = Actor::new(signer); + + let a1 = alice.comment("First comment", None); + let a2 = alice.comment("Second comment", None); + + let mut expected = Thread::default(); + expected.apply([a1.clone(), a2.clone()]); + + let created = store + .create("Thread created", a1.encode(), &alice.signer) + .unwrap(); + store + .update(*created.id(), "Thread updated", a2.encode(), &alice.signer) + .unwrap(); + + let actual = store.get(created.id()).unwrap().unwrap(); + + assert_eq!(actual, expected); + } + + #[test] + fn test_timelines_basic() { + let mut alice = Actor::::default(); + let mut bob = Actor::::default(); + + let a1 = alice.comment("First comment", None); + let a2 = alice.comment("Second comment", None); + + bob.receive([a1.clone(), a2.clone()]); + assert_eq!( + bob.timeline().collect::>(), + alice.timeline().collect::>() + ); + assert_eq!(alice.timeline().collect::>(), vec![&a1, &a2]); + + bob.reset(); + bob.receive([a2, a1]); + assert_eq!( + bob.timeline().collect::>(), + alice.timeline().collect::>() + ); + } + + #[test] + fn test_timelines_concurrent() { + let mut alice = Actor::::default(); + let mut bob = Actor::::default(); + let mut eve = Actor::::default(); + + let a1 = alice.comment("First comment", None); + + bob.receive([a1.clone()]); + + let b0 = bob.comment("Bob's first reply to Alice", None); + let b1 = bob.comment("Bob's second reply to Alice", None); + + eve.receive([b1.clone(), b0.clone()]); + let e0 = eve.comment("Eve's first reply to Alice", None); + + bob.receive([e0.clone()]); + let b2 = bob.comment("Bob's third reply to Alice", None); + + eve.receive([b2.clone(), a1.clone()]); + let e1 = eve.comment("Eve's second reply to Alice", None); + + alice.receive([b0.clone(), b1.clone(), b2.clone(), e0.clone(), e1.clone()]); + bob.receive([e1.clone()]); + + let a2 = alice.comment("Second comment", None); + eve.receive([a2.clone()]); + bob.receive([a2.clone()]); + + assert_eq!(alice.changes.len(), 7); + assert_eq!(bob.changes.len(), 7); + assert_eq!(eve.changes.len(), 7); + + assert_eq!( + bob.timeline().collect::>(), + alice.timeline().collect::>() + ); + assert_eq!( + eve.timeline().collect::>(), + alice.timeline().collect::>() + ); + assert_eq!( + vec![&a1, &b0, &b1, &e0, &b2, &e1, &a2], + alice.timeline().collect::>(), + ); + } + + #[quickcheck] + fn prop_invariants(log: Changes<3>) { + let t = Thread::default(); + let [p1, p2, p3] = log.permutations; + + let mut t1 = t.clone(); + t1.apply(p1); + + let mut t2 = t.clone(); + t2.apply(p2); + + let mut t3 = t; + t3.apply(p3); + + assert_eq!(t1, t2); + assert_eq!(t2, t3); + assert_laws(&t1, &t2, &t3); + } +}