cob: Unify change decoding
Signed-off-by: Alexis Sellier <alexis@radicle.xyz>
This commit is contained in:
parent
4e0efa4690
commit
fa3c155865
|
|
@ -1,15 +1,25 @@
|
||||||
use std::collections::BTreeMap;
|
use std::collections::BTreeMap;
|
||||||
|
|
||||||
|
use serde::{Deserialize, Serialize};
|
||||||
|
use thiserror::Error;
|
||||||
|
|
||||||
|
use radicle_cob::history::EntryWithClock;
|
||||||
use radicle_crdt::clock;
|
use radicle_crdt::clock;
|
||||||
use radicle_crdt::clock::Lamport;
|
use radicle_crdt::clock::Lamport;
|
||||||
use radicle_crypto::{PublicKey, Signer};
|
use radicle_crypto::{PublicKey, Signer};
|
||||||
use serde::{Deserialize, Serialize};
|
|
||||||
|
|
||||||
/// Identifies a change.
|
/// Identifies a change.
|
||||||
pub type ChangeId = (Lamport, ActorId);
|
pub type ChangeId = (Lamport, ActorId);
|
||||||
/// The author of a change.
|
/// The author of a change.
|
||||||
pub type ActorId = PublicKey;
|
pub type ActorId = PublicKey;
|
||||||
|
|
||||||
|
/// Error decoding a change from an entry.
|
||||||
|
#[derive(Error, Debug)]
|
||||||
|
pub enum ChangeDecodeError {
|
||||||
|
#[error("deserialization from json failed: {0}")]
|
||||||
|
Deserialize(#[from] serde_json::Error),
|
||||||
|
}
|
||||||
|
|
||||||
/// The `Change` is the unit of replication.
|
/// The `Change` is the unit of replication.
|
||||||
/// Everything that can be done in the system is represented by a `Change` object.
|
/// 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.
|
/// Changes are applied to an accumulator to yield a final state.
|
||||||
|
|
@ -25,6 +35,21 @@ pub struct Change<A> {
|
||||||
pub timestamp: clock::Physical,
|
pub timestamp: clock::Physical,
|
||||||
}
|
}
|
||||||
|
|
||||||
|
impl<'a: 'de, 'de, A: serde::Deserialize<'de>> TryFrom<&'a EntryWithClock> for Change<A> {
|
||||||
|
type Error = ChangeDecodeError;
|
||||||
|
|
||||||
|
fn try_from(entry: &'a EntryWithClock) -> Result<Self, Self::Error> {
|
||||||
|
let action = serde_json::from_slice(entry.contents())?;
|
||||||
|
|
||||||
|
Ok(Change {
|
||||||
|
action,
|
||||||
|
author: *entry.actor(),
|
||||||
|
clock: entry.clock().into(),
|
||||||
|
timestamp: entry.timestamp().into(),
|
||||||
|
})
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
impl<A> Change<A> {
|
impl<A> Change<A> {
|
||||||
/// Get the change id.
|
/// Get the change id.
|
||||||
pub fn id(&self) -> ChangeId {
|
pub fn id(&self) -> ChangeId {
|
||||||
|
|
|
||||||
|
|
@ -111,13 +111,8 @@ impl store::FromHistory for Issue {
|
||||||
history: &radicle_cob::History,
|
history: &radicle_cob::History,
|
||||||
) -> Result<(Self, clock::Lamport), store::Error> {
|
) -> Result<(Self, clock::Lamport), store::Error> {
|
||||||
let obj = history.traverse(Self::default(), |mut acc, entry| {
|
let obj = history.traverse(Self::default(), |mut acc, entry| {
|
||||||
if let Ok(action) = Action::decode(entry.contents()) {
|
if let Ok(change) = Change::try_from(entry) {
|
||||||
if let Err(err) = acc.apply(Change {
|
if let Err(err) = acc.apply(change) {
|
||||||
action,
|
|
||||||
author: *entry.actor(),
|
|
||||||
clock: entry.clock().into(),
|
|
||||||
timestamp: entry.timestamp().into(),
|
|
||||||
}) {
|
|
||||||
log::warn!("Error applying change to issue state: {err}");
|
log::warn!("Error applying change to issue state: {err}");
|
||||||
return ControlFlow::Break(acc);
|
return ControlFlow::Break(acc);
|
||||||
}
|
}
|
||||||
|
|
@ -394,12 +389,6 @@ pub enum Action {
|
||||||
Thread { action: thread::Action },
|
Thread { action: thread::Action },
|
||||||
}
|
}
|
||||||
|
|
||||||
impl Action {
|
|
||||||
pub fn decode(bytes: &[u8]) -> Result<Self, serde_json::Error> {
|
|
||||||
serde_json::from_slice(bytes)
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
impl From<thread::Action> for Action {
|
impl From<thread::Action> for Action {
|
||||||
fn from(action: thread::Action) -> Self {
|
fn from(action: thread::Action) -> Self {
|
||||||
Self::Thread { action }
|
Self::Thread { action }
|
||||||
|
|
|
||||||
|
|
@ -99,12 +99,6 @@ pub enum Action {
|
||||||
},
|
},
|
||||||
}
|
}
|
||||||
|
|
||||||
impl Action {
|
|
||||||
pub fn decode(bytes: &[u8]) -> Result<Self, serde_json::Error> {
|
|
||||||
serde_json::from_slice(bytes)
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
/// Where a patch is intended to be merged.
|
/// Where a patch is intended to be merged.
|
||||||
#[derive(Default, Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Serialize, Deserialize)]
|
#[derive(Default, Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Serialize, Deserialize)]
|
||||||
#[serde(rename_all = "lowercase")]
|
#[serde(rename_all = "lowercase")]
|
||||||
|
|
@ -332,13 +326,8 @@ impl store::FromHistory for Patch {
|
||||||
history: &radicle_cob::History,
|
history: &radicle_cob::History,
|
||||||
) -> Result<(Self, clock::Lamport), store::Error> {
|
) -> Result<(Self, clock::Lamport), store::Error> {
|
||||||
let obj = history.traverse(Self::default(), |mut acc, entry| {
|
let obj = history.traverse(Self::default(), |mut acc, entry| {
|
||||||
if let Ok(action) = Action::decode(entry.contents()) {
|
if let Ok(change) = Change::try_from(entry) {
|
||||||
if let Err(err) = acc.apply([Change {
|
if let Err(err) = acc.apply([change]) {
|
||||||
action,
|
|
||||||
author: *entry.actor(),
|
|
||||||
clock: entry.clock().into(),
|
|
||||||
timestamp: entry.timestamp().into(),
|
|
||||||
}]) {
|
|
||||||
log::warn!("Error applying change to patch state: {err}");
|
log::warn!("Error applying change to patch state: {err}");
|
||||||
return ControlFlow::Break(acc);
|
return ControlFlow::Break(acc);
|
||||||
}
|
}
|
||||||
|
|
|
||||||
|
|
@ -78,13 +78,6 @@ pub enum Action {
|
||||||
},
|
},
|
||||||
}
|
}
|
||||||
|
|
||||||
impl Action {
|
|
||||||
/// Deserialize an action from a byte string.
|
|
||||||
pub fn decode(bytes: &[u8]) -> Result<Self, serde_json::Error> {
|
|
||||||
serde_json::from_slice(bytes)
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
/// A discussion thread.
|
/// A discussion thread.
|
||||||
#[derive(Debug, Default, Clone, PartialEq, Eq)]
|
#[derive(Debug, Default, Clone, PartialEq, Eq)]
|
||||||
pub struct Thread {
|
pub struct Thread {
|
||||||
|
|
@ -103,13 +96,8 @@ impl store::FromHistory for Thread {
|
||||||
|
|
||||||
fn from_history(history: &History) -> Result<(Self, Lamport), store::Error> {
|
fn from_history(history: &History) -> Result<(Self, Lamport), store::Error> {
|
||||||
let obj = history.traverse(Thread::default(), |mut acc, entry| {
|
let obj = history.traverse(Thread::default(), |mut acc, entry| {
|
||||||
if let Ok(action) = Action::decode(entry.contents()) {
|
if let Ok(change) = Change::try_from(entry) {
|
||||||
acc.apply([Change {
|
acc.apply([change]);
|
||||||
action,
|
|
||||||
author: *entry.actor(),
|
|
||||||
clock: entry.clock().into(),
|
|
||||||
timestamp: entry.timestamp().into(),
|
|
||||||
}]);
|
|
||||||
ControlFlow::Continue(acc)
|
ControlFlow::Continue(acc)
|
||||||
} else {
|
} else {
|
||||||
ControlFlow::Break(acc)
|
ControlFlow::Break(acc)
|
||||||
|
|
|
||||||
Loading…
Reference in New Issue