Refactor and create worker target struct

Change-type: minor

Cargo.toml

@@ -5,12 +5,12 @@ edition = "2021"
 description = "An automated job orchestration library to build and execute dynamic workflows"
 
 [dependencies]
-json-patch = "2.0.0"
+json-patch = "3"
 jsonptr = "0.6.0"
 matchit = "0.8.4"
 serde = { version = "1.0.197" }
 serde_json = "1.0.120"
-thiserror = "1.0.63"
+thiserror = "2"
 
 [dev-dependencies]
 tokio = { version = "1.36.0", features = ["rt", "macros", "time"] }

src/extract/path/error.rs

@@ -89,10 +89,6 @@ impl fmt::Display for PathDeserializationError {
 impl std::error::Error for PathDeserializationError {}
 
 /// The kinds of errors that can happen we deserializing into a [`Path`].
-///
-/// This type is obtained through [`FailedToDeserializePathParams::kind`] or
-/// [`FailedToDeserializePathParams::into_kind`] and is useful for building
-/// more precise error messages.
 #[derive(Debug, PartialEq, Eq)]
 #[non_exhaustive]
 pub enum ErrorKind {

src/lib.rs

@@ -4,3 +4,4 @@ pub mod error;
 pub mod extract;
 pub mod system;
 pub mod task;
+pub mod worker;

src/system/mod.rs

@@ -1,7 +1,10 @@
 use json_patch::{patch, Patch};
 use serde::{de::DeserializeOwned, Serialize};
 use serde_json::Value;
-use std::fmt::{self, Display};
+use std::{
+    fmt::{self, Display},
+    ops::Deref,
+};
 
 use crate::error::{Error, IntoError};
 
@@ -82,3 +85,11 @@ impl System {
         Ok(s)
     }
 }
+
+impl Deref for System {
+    type Target = Value;
+
+    fn deref(&self) -> &Self::Target {
+        &self.state
+    }
+}

src/task/mod.rs

@@ -1,9 +1,7 @@
 mod boxed;
 mod context;
-mod domain;
 mod effect;
 mod handler;
-mod intent;
 mod job;
 mod result;
 
@@ -16,10 +14,8 @@ use crate::error::{Error, IntoError};
 use crate::system::System;
 
 pub use context::*;
-pub use domain::*;
 pub use effect::*;
 pub use handler::*;
-pub use intent::*;
 pub use job::*;
 pub(crate) use result::*;
 
@@ -133,6 +129,7 @@ impl<S> Task<S> {
 
     /// Run every action in the task sequentially and return the
     /// aggregate changes.
+    /// TODO: this should probably only have crate visibility
     pub fn dry_run(self, system: &System) -> Result<Patch> {
         match self {
             Self::Atom {

src/task/domain.rs → src/worker/domain.rs

@@ -133,11 +133,13 @@ impl<S> Domain<S> {
 
 #[cfg(test)]
 mod tests {
+    use super::*;
     use std::sync::Arc;
 
     use crate::extract::{Target, Update};
     use crate::path::PathArgs;
     use crate::task::*;
+    use crate::worker::*;
 
     fn plus_one(mut counter: Update<i32>, tgt: Target<i32>) -> Update<i32> {
         if *counter < *tgt {

src/task/intent.rs → src/worker/intent.rs

@@ -1,9 +1,8 @@
-use super::handler::Handler;
-use super::job::Job;
+use crate::task::{Handler, Job};
 use std::cmp::Ordering;
 
 #[derive(PartialEq, PartialOrd, Eq, Ord, Debug, Clone)]
-pub(crate) enum Operation {
+pub enum Operation {
     Create,
     Update,
     Delete,
@@ -39,7 +38,7 @@ impl<S> Intent<S> {
         }
     }
 
-    pub(crate) fn with_operation(self, operation: Operation) -> Self {
+    fn with_operation(self, operation: Operation) -> Self {
         let Intent { priority, job, .. } = self;
         Intent {
             operation,

src/worker/mod.rs

@@ -0,0 +1,7 @@
+mod domain;
+mod intent;
+mod target;
+
+pub use domain::*;
+pub use intent::*;
+pub use target::*;

src/worker/target.rs

@@ -0,0 +1,246 @@
+use json_patch::{diff, Patch, PatchOperation, RemoveOperation, ReplaceOperation};
+use jsonptr::Pointer;
+use serde::Serialize;
+use serde_json::Value;
+use std::{
+    cmp::Ordering,
+    collections::{BTreeSet, LinkedList},
+    fmt::{self, Display},
+    ops::Deref,
+};
+
+pub struct Target(Value);
+
+#[derive(Debug)]
+pub enum TargetError {
+    SerializationFailed(serde_json::error::Error),
+}
+
+impl std::error::Error for TargetError {}
+
+impl Display for TargetError {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        match self {
+            TargetError::SerializationFailed(err) => err.fmt(f),
+        }
+    }
+}
+
+#[derive(PartialEq, Eq, Debug, Clone)]
+pub(crate) struct Operation(PatchOperation);
+
+impl PartialOrd for Operation {
+    fn partial_cmp(&self, other: &Self) -> Option<Ordering> {
+        Some(self.cmp(other))
+    }
+}
+
+impl Ord for Operation {
+    fn cmp(&self, other: &Self) -> Ordering {
+        let thispath = self.0.path();
+        let otherpath = other.0.path();
+
+        // Order operations by path length
+        thispath
+            .count()
+            .cmp(&otherpath.count())
+            .then(thispath.cmp(otherpath))
+    }
+}
+
+impl From<PatchOperation> for Operation {
+    fn from(op: PatchOperation) -> Operation {
+        Operation(op)
+    }
+}
+
+pub(crate) struct Distance(BTreeSet<Operation>);
+
+impl Distance {
+    fn new() -> Self {
+        Distance(BTreeSet::new())
+    }
+
+    fn insert(&mut self, o: Operation) {
+        self.0.insert(o);
+    }
+
+    fn iter(&self) -> impl Iterator<Item = &Operation> {
+        self.0.iter()
+    }
+
+    fn add_removes_for_children(&mut self, path: &Pointer, value: &Value) {
+        let mut queue = LinkedList::new();
+        queue.push_back((path.to_buf(), value));
+
+        while let Some((path, value)) = queue.pop_front() {
+            if value.is_object() {
+                let obj = value.as_object().unwrap();
+                for (k, v) in obj.iter() {
+                    let path = path.concat(Pointer::parse(&format!("/{}", k)).unwrap());
+                    // Insert a replace operation for each child
+                    self.insert(Operation::from(PatchOperation::Remove(RemoveOperation {
+                        path: path.clone(),
+                    })));
+
+                    // Append the value to the queue
+                    queue.push_back((path, v));
+                }
+            }
+
+            if value.is_array() {
+                let obj = value.as_array().unwrap();
+                for (k, v) in obj.iter().enumerate() {
+                    let path = path.concat(Pointer::parse(&format!("/{}", k)).unwrap());
+
+                    // Insert a replace operation for each child
+                    self.insert(Operation::from(PatchOperation::Remove(RemoveOperation {
+                        path: path.clone(),
+                    })));
+
+                    // Append the value to the queue
+                    queue.push_back((path, v));
+                }
+            }
+        }
+    }
+}
+
+impl Target {
+    pub fn new(target: impl Into<Value>) -> Self {
+        Self(target.into())
+    }
+
+    pub fn from<S: Serialize>(state: S) -> Result<Self, TargetError> {
+        let state = serde_json::to_value(state).map_err(TargetError::SerializationFailed)?;
+        Ok(Target::new(state))
+    }
+
+    pub(crate) fn distance(&self, state: &Value) -> Distance {
+        let mut distance = Distance::new();
+
+        // calculate differences between the system root and
+        // the target
+        let Patch(changes) = diff(state, self);
+        for op in changes {
+            // For every operation on the list of changes
+            let path = op.path();
+
+            let mut parent = path;
+
+            // get all paths up to the root
+            while let Some(newparent) = parent.parent() {
+                // get the target at the parent to use as value
+                // no matter the operation, the parent of the target shoul
+                // always exist. If it doesn't there is a bug (probbly in jsonptr)
+                let value = newparent.resolve(&self.0).unwrap_or_else(|e| {
+                    panic!(
+                        "[BUG] Path `{}` should be resolvable on the target, but got error: {}",
+                        newparent, e
+                    )
+                });
+
+                // Insert a replace operation for each one
+                distance.insert(Operation::from(PatchOperation::Replace(ReplaceOperation {
+                    path: newparent.to_buf(),
+                    value: value.clone(),
+                })));
+
+                parent = newparent;
+            }
+
+            // for every delete operation '/a/b/c', add child
+            // nodes of the deleted path with 'remove' operation
+            if let PatchOperation::Remove(_) = op {
+                // By definition of the remove operation the path should be
+                // resolvable on the left side of the diff
+                let value = path.resolve(state).unwrap_or_else(|e| {
+                    panic!(
+                        "[BUG] Path `{}` should be resolvable on the state, but got error: {}",
+                        path, e
+                    )
+                });
+
+                //
+                distance.add_removes_for_children(path, value);
+            }
+
+            // Finally insert the actual operation
+            distance.insert(Operation::from(op));
+        }
+
+        distance
+    }
+}
+
+impl Deref for Target {
+    type Target = Value;
+
+    fn deref(&self) -> &Self::Target {
+        &self.0
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use serde_json::json;
+
+    fn distance_eq(src: Value, tgt: Value, result: Vec<Value>) {
+        let target = Target::new(tgt);
+        let distance = target.distance(&src);
+
+        // Serialize results to make comparison easier
+        let ops: Vec<Value> = distance
+            .iter()
+            .map(|Operation(o)| serde_json::to_value(o).unwrap())
+            .collect();
+
+        assert_eq!(ops, result)
+    }
+
+    #[test]
+    fn calculates_possible_changes_to_target() {
+        distance_eq(
+            json!({"a": 1, "b": "one", "c": {"k": "v"}}),
+            json!({"a": 2, "b": "one", "c": {}}),
+            vec![
+                json!({"op": "replace", "path": "", "value": {"a": 2, "b": "one", "c": {}}}),
+                json!({"op": "replace", "path": "/a", "value": 2}),
+                json!({"op": "replace", "path": "/c", "value": {}}),
+                json!({"op": "remove", "path": "/c/k"}),
+            ],
+        );
+        distance_eq(
+            json!({"a": 1, "b": "one", "c": {"k": "v"}}),
+            json!({"a": 2}),
+            vec![
+                json!({"op": "replace", "path": "", "value": {"a": 2}}),
+                json!({"op": "replace", "path": "/a", "value": 2}),
+                json!({"op": "remove", "path": "/b"}),
+                json!({"op": "remove", "path": "/c"}),
+                json!({"op": "remove", "path": "/c/k"}),
+            ],
+        );
+        distance_eq(
+            json!({"a": 1, "b": "one", "c": {"k": "v"}}),
+            json!({"a": 2, "b": "two", "c": {"k": "v"}}),
+            vec![
+                json!({"op": "replace", "path": "", "value": {"a": 2, "b": "two", "c": {"k": "v"}}}),
+                json!({"op": "replace", "path": "/a", "value": 2}),
+                json!({"op": "replace", "path": "/b", "value": "two"}),
+            ],
+        );
+        distance_eq(
+            json!({"a": {"b": {"c": {"d": "e"}}}}),
+            json!({"a": {"b": {}}}),
+            vec![
+                json!({"op": "replace", "path": "", "value": {"a": {"b": {}}}}),
+                json!({"op": "replace", "path": "/a", "value": {"b": {}}}),
+                json!({"op": "replace", "path": "/a/b", "value": {}}),
+                json!({"op": "remove", "path": "/a/b/c"}),
+                json!({"op": "remove", "path": "/a/b/c/d"}),
+            ],
+        );
+    }
+}