add cron scheduler to moose (#1742)

Author: DatGuyJonathan

Cargo.lock

@@ -77,6 +77,7 @@ pbkdf2 = { version = "0.12", features = ["simple"] }
 sha2 = "0.10.8"
 hex = "0.4.2"
 constant_time_eq = "0.3.0"
+tokio-cron-scheduler = "0.11.0"
 
 [dev-dependencies]
 clickhouse = { version = "0.11.5", features = ["uuid", "test-util"] }

apps/framework-cli/Cargo.toml

@@ -16,6 +16,7 @@ use super::olap::clickhouse::{errors::ClickhouseError, mapper::std_columns_to_cl
 
 pub mod aggregations_registry;
 pub mod consumption_registry;
+pub mod cron_registry;
 pub mod functions_registry;
 pub mod kafka_clickhouse_sync;
 pub mod process_registry;

apps/framework-cli/src/infrastructure/processes.rs

@@ -0,0 +1,56 @@
+use std::sync::Arc;
+use tokio::sync::Mutex;
+use tokio::task;
+use tokio_cron_scheduler::{Job, JobScheduler, JobSchedulerError};
+
+pub struct CronRegistry {
+    scheduler: Arc<Mutex<JobScheduler>>,
+}
+
+impl CronRegistry {
+    pub async fn new() -> Result<Self, JobSchedulerError> {
+        let scheduler = JobScheduler::new().await?;
+
+        Ok(CronRegistry {
+            scheduler: Arc::new(Mutex::new(scheduler)),
+        })
+    }
+
+    pub async fn add_job<F>(&self, cron_expression: &str, task: F) -> Result<(), JobSchedulerError>
+    where
+        F: Fn() -> Result<(), String> + Send + Sync + 'static,
+    {
+        let task = Arc::new(task);
+
+        let job = Job::new_async(cron_expression, move |_uuid, _l| {
+            let task = task.clone();
+            Box::pin(async move {
+                let _ = task();
+            })
+        })?;
+
+        let scheduler = self.scheduler.lock().await;
+        scheduler.add(job).await?;
+
+        Ok(())
+    }
+
+    pub async fn start(&self) -> Result<(), JobSchedulerError> {
+        let scheduler: Arc<Mutex<JobScheduler>> = self.scheduler.clone();
+        task::spawn(async move {
+            let scheduler = scheduler.lock().await;
+            scheduler.start().await?;
+
+            Ok::<(), JobSchedulerError>(())
+        });
+
+        Ok(())
+    }
+
+    pub async fn stop(&self) -> Result<(), JobSchedulerError> {
+        let mut scheduler = self.scheduler.lock().await;
+        scheduler.shutdown().await?;
+
+        Ok(())
+    }
+}