chore: add pause and resume feature in workflow management

actions/workflow-invoker/src/lib.rs

@@ -2,6 +2,7 @@ extern crate serde_json;
 
 use serde_derive::{Deserialize, Serialize};
 use serde_json::{Error, Value};
+use types::UserData;
 mod types;
 use crate::types::update_with;
 #[cfg(test)]
@@ -56,26 +57,30 @@ impl Action {
         self.context.as_mut().expect("Action not Initialized!")
     }
 
-    pub fn fetch_input(&mut self) -> Result<Vec<Value>, Error> {
+    pub fn fetch_input(&mut self) -> Result<Vec<UserData>, Error> {
         let id = self.params.messages.clone()[0].topic.clone();
         let data = self.get_context().get_document(&id)?;
         let parsed = serde_json::from_value::<Topic>(data)?;
         Ok(parsed.data)
     }
 
-    pub fn invoke_trigger(&mut self, payload: &mut Vec<Value>) -> Result<Value, Error> {
+    pub fn invoke_trigger(&mut self, payload: &mut Vec<UserData>) -> Result<Value, Error> {
         let mut failed_triggers = vec![];
-        for message in payload.iter_mut() {
+
+        for user in payload.iter_mut() {
             let data = serde_json::from_str::<Value>(&self.params.messages[0].value).unwrap();
-            update_with(message, &data);
+            update_with(&mut user.input_data, &data);
 
             let trigger = self.params.polkadot_payout_trigger.clone();
-            if self
-                .get_context()
-                .invoke_trigger(&trigger, &serde_json::json!({"data": message}))
-                .is_err()
-            {
-                failed_triggers.push(self.params.messages[0].value.clone());
+
+            if user.status == "active".to_string() {
+                if self
+                    .get_context()
+                    .invoke_trigger(&trigger, &serde_json::json!({"data": user.input_data}))
+                    .is_err()
+                {
+                    failed_triggers.push(self.params.messages[0].value.clone());
+                }
             }
         }
         if !failed_triggers.is_empty() {
@@ -132,7 +137,11 @@ mod tests {
         let workflow_db = action.connect_db(&action.params.db_url, &action.params.db_name);
         let workflow_management_db_context = Context::new(workflow_db, None);
         let doc = serde_json::json!({
-            "data": [{ "url": "todo!()".to_string(), "validator": "todo!()".to_string(), "owner_key": "todo!()".to_string() }]
+            "data": [{
+                "user_id" : "asdf",
+                "status" : "active",
+                "input_data" :{ "url": "todo!()".to_string(), "validator": "todo!()".to_string(), "owner_key": "todo!()".to_string() }
+            }]
         });
         let _ = workflow_management_db_context
             .insert_document(&doc, Some(action.params.messages[0].topic.clone()));

actions/workflow-invoker/src/types/data.rs

@@ -7,3 +7,20 @@ pub struct DbDatas {
     pub validator: String,
     pub key: String,
 }
+
+#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
+pub struct Topic {
+    #[serde(rename = "_id")]
+    pub id: String,
+    #[serde(rename = "_rev")]
+    pub rev: String,
+    pub data: Vec<UserData>,
+}
+
+#[allow(clippy::derive_partial_eq_without_eq)]
+#[derive(Debug, Clone, PartialEq, Serialize, Deserialize, Default)]
+pub struct UserData {
+    pub user_id: String,
+    pub status: String,
+    pub input_data: serde_json::Value,
+}

actions/workflow-invoker/src/types/mod.rs

@@ -3,9 +3,8 @@ pub mod source;
 pub mod topic;
 pub use message::{Era, Message};
 pub use source::Source;
-pub use topic::Topic;
 mod data;
-pub use data::*;
+pub use data::{Topic, UserData};
 
 pub fn update_with(dest: &mut serde_json::Value, src: &serde_json::Value) {
     use serde_json::Value::{Null, Object};

actions/workflow-invoker/src/types/topic.rs

@@ -1,14 +1,4 @@
 use serde_derive::{Deserialize, Serialize};
-use serde_json::Value;
-
-#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
-pub struct Topic {
-    #[serde(skip_serializing, rename(deserialize = "_id"))]
-    pub id: String,
-    #[serde(skip_serializing, rename(deserialize = "_rev"))]
-    pub rev: String,
-    pub data: Vec<Value>,
-}
 
 #[allow(clippy::derive_partial_eq_without_eq)]
 #[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]

actions/workflow_management/Cargo.toml

@@ -33,7 +33,10 @@ actions-common = { git = "https://github.com/HugoByte/aurras", rev = '1f7e117' }
 reqwest = { version = "0.11", features = ["blocking", "json"] }
 bcrypt = "0.13.0"
 jsonwebtoken = "7.1"
+chrono = { version = "0.4", features = ["serde"] }
 
 [dev-dependencies]
-actions-common = { git = "https://github.com/HugoByte/aurras", rev = '1f7e117', features = ["mock_containers"] }
-tokio = { version = "1.0.0", features = ["macros"] }
+actions-common = { git = "https://github.com/HugoByte/aurras", rev = '1f7e117', features = [
+    "mock_containers",
+] }
+tokio = { version = "1.0.0", features = ["macros"] }