feat: implemented cron matching logic

Author: raphael-goetz

adapter/cron/src/main.rs

@@ -1,9 +1,10 @@
 use std::str::FromStr;
 use async_trait::async_trait;
-use chrono::{Local, Utc};
+use chrono::{DateTime, Datelike, Timelike, Utc};
 use cron::Schedule;
 use base::extract_flow_setting_field;
 use base::runner::{ServerContext, ServerRunner};
+use base::store::FlowIdentifyResult;
 use base::traits::{IdentifiableFlow, LoadConfig, Server};
 
 #[derive(Default)]
@@ -27,15 +28,15 @@ async fn main() {
 }
 
 struct Time {
-    utc: Utc
+    now: DateTime<Utc>
 }
 
 impl IdentifiableFlow for Time {
     fn identify(&self, flow: &tucana::shared::ValidationFlow) -> bool {
         let Some(minute) = extract_flow_setting_field(&flow.settings, "CRON_MINUTE", "minute") else {
             return false;
         };
-        let Some(hour) = extract_flow_setting_field(&flow.settings, "CRON_MOUR", "hour") else {
+        let Some(hour) = extract_flow_setting_field(&flow.settings, "CRON_HOUR", "hour") else {
             return false;
         };
         let Some(dom) = extract_flow_setting_field(&flow.settings, "CRON_DAY_OF_MONTH", "day_of_month") else {
@@ -48,12 +49,15 @@ impl IdentifiableFlow for Time {
             return false;
         };
 
-        let expression = format!("{} {} {} {} {}", minute, hour, dom, month, dow);
+        let expression = format!("* {} {} {} {} {}", minute, hour, dom, month, dow);
+        let schedule = Schedule::from_str(expression.as_str()).unwrap();
+        let next = schedule.upcoming(Utc).next().unwrap();
 
-
-
-
-        todo!()
+        self.now.year() == next.year() &&
+            self.now.month() == next.month() &&
+            self.now.day() == next.day() &&
+            self.now.hour() == next.hour() &&
+            self.now.minute() == next.minute()
     }
 }
 
@@ -66,19 +70,28 @@ impl Server<CronConfig> for Cron {
     async fn run(&mut self, ctx: &ServerContext<CronConfig>) -> anyhow::Result<()> {
         let expression = "0 * * * * *";
         let schedule = Schedule::from_str(expression).expect("Failed to parse CRON expression");
+        let pattern = "*.*.CRON.*";
 
         loop {
             let now = Utc::now();
             if let Some(next) = schedule.upcoming(Utc).take(1).next() {
                 let until_next = next - now;
                 tokio::time::sleep(until_next.to_std()?).await;
-                println!(
-                    "Running every minute. Current time: {}",
-                    Local::now().format("%Y-%m-%d %H:%M:%S")
-                );
+
+                let time = Time { now };
+                match ctx.adapter_store.get_possible_flow_match(pattern.to_string(), time).await {
+                    FlowIdentifyResult::None => {}
+                    FlowIdentifyResult::Single(flow) => {
+                        ctx.adapter_store.validate_and_execute_flow(flow, None, false).await;
+                    }
+                    FlowIdentifyResult::Multiple(flows) => {
+                        for flow in flows {
+                            ctx.adapter_store.validate_and_execute_flow(flow, None, false).await;
+                        }
+                    },
+                }
             }
         }
-        Ok(())
     }
 
     async fn shutdown(&mut self, _ctx: &ServerContext<CronConfig>) -> anyhow::Result<()> {

adapter/rest/src/main.rs

@@ -96,7 +96,7 @@ async fn execute_flow(
     request: HttpRequest,
     store: Arc<base::store::AdapterStore>,
 ) -> Option<HttpResponse> {
-    match store.validate_and_execute_flow(flow, request.body).await {
+    match store.validate_and_execute_flow(flow, request.body, true).await {
         Some(result) => {
             let Value {
                 kind: Some(StructValue(Struct { fields })),

crates/base/src/store.rs

@@ -101,10 +101,34 @@ impl AdapterStore {
         }
     }
 
-    pub async fn execute(&self, flow: ExecutionFlow, wait_for_result: bool) -> Option<Value> {
+    /// validate_and_execute_flow
+    ///
+    /// This function will validate the flow. If the flow is valid, it will execute (send the flow to the execution and wait for a/multiple result/s) the flow.
+    ///
+    /// Arguments:
+    /// - flow: The flow to be validated and executed.
+    /// - input_value: The input value to be used for the flow execution.
+    pub async fn validate_and_execute_flow(
+        &self,
+        flow: ValidationFlow,
+        input_value: Option<Value>,
+        wait_for_result: bool,
+    ) -> Option<Value> {
+        if let Some(body) = input_value.clone() {
+            let verify_result = verify_flow(flow.clone(), body);
+
+            match verify_result {
+                Ok(()) => {}
+                Err(_err) => {
+                    return None;
+                }
+            };
+        }
+
+        let execution_flow: ExecutionFlow = Self::convert_validation_flow(flow, input_value);
         let uuid = uuid::Uuid::new_v4().to_string();
         let topic = format!("execution.{}", uuid);
-        let bytes = flow.encode_to_vec();
+        let bytes = execution_flow.encode_to_vec();
 
         if wait_for_result {
             match self.client.request(topic, bytes.into()).await {
@@ -131,33 +155,6 @@ impl AdapterStore {
         }
     }
 
-    /// validate_and_execute_flow
-    ///
-    /// This function will validate the flow. If the flow is valid, it will execute (send the flow to the execution and wait for a/multiple result/s) the flow.
-    ///
-    /// Arguments:
-    /// - flow: The flow to be validated and executed.
-    /// - input_value: The input value to be used for the flow execution.
-    pub async fn validate_and_execute_flow(
-        &self,
-        flow: ValidationFlow,
-        input_value: Option<Value>,
-    ) -> Option<Value> {
-        if let Some(body) = input_value.clone() {
-            let verify_result = verify_flow(flow.clone(), body);
-
-            match verify_result {
-                Ok(()) => {}
-                Err(_err) => {
-                    return None;
-                }
-            };
-        }
-
-        let execution_flow: ExecutionFlow = Self::convert_validation_flow(flow, input_value);
-        self.execute(execution_flow, true).await
-    }
-
     fn convert_validation_flow(flow: ValidationFlow, input_value: Option<Value>) -> ExecutionFlow {
         ExecutionFlow {
             flow_id: flow.flow_id,