Cron jobs support in main runner

Author: Mauricio Cassola

Cargo.lock

@@ -42,6 +42,7 @@ tower = { version = "0.4.13", features = ["util", "limit", "buffer", "load-shed"
 github-graphql = { path = "github-graphql" }
 rand = "0.8.5"
 ignore = "0.4.18"
+postgres-types = { version = "0.2.4", features = ["derive"] }
 
 [dependencies.serde]
 version = "1"

Cargo.toml

@@ -195,6 +195,15 @@ pub async fn run_scheduled_jobs(db: &DbClient) -> anyhow::Result<()> {
                 println!("job succesfully executed (id={})", job.id);
                 tracing::trace!("job succesfully executed (id={})", job.id);
 
+                if let Some(frequency) = job.frequency {
+                    let duration = get_duration_from_cron(frequency, job.frequency_unit.as_ref().unwrap());
+                    let new_expected_time = job.expected_time.checked_add_signed(duration).unwrap();
+
+                    insert_job(&db, &job.name, &new_expected_time, &Some(frequency), &job.frequency_unit, &job.metadata).await?;
+                    println!("job succesfully reinserted (name={})", job.name);
+                    tracing::trace!("job succesfully reinserted (name={})", job.name);
+                }
+
                 delete_job(&db, &job.id).await?;
             },
             Err(e) => {
@@ -245,12 +254,17 @@ CREATE TABLE issue_data (
     data JSONB,
     PRIMARY KEY (repo, issue_number, key)
 );
+",
+"
+CREATE TYPE frequency_unit AS ENUM ('days', 'hours', 'minutes', 'seconds');
 ",
     "
 CREATE TABLE jobs (
     id UUID DEFAULT gen_random_uuid() PRIMARY KEY,
     name TEXT NOT NULL,
     expected_time TIMESTAMP WITH TIME ZONE NOT NULL,
+    frequency INTEGER,
+    frequency_unit frequency_unit,
     metadata JSONB,
     executed_at TIMESTAMP WITH TIME ZONE,
     error_message TEXT

src/db.rs

@@ -1,32 +1,50 @@
 //! The `jobs` table provides a way to have scheduled jobs
 use anyhow::{Result, Context as _};
-use chrono::{DateTime, FixedOffset};
+use chrono::{DateTime, FixedOffset, Duration};
 use tokio_postgres::{Client as DbClient};
 use uuid::Uuid;
 use serde::{Deserialize, Serialize};
+use postgres_types::{ToSql, FromSql};
 
 #[derive(Serialize, Deserialize, Debug)]
 pub struct Job {
     pub id: Uuid,
     pub name: String,
     pub expected_time: DateTime<FixedOffset>,
+    pub frequency: Option<i32>,
+    pub frequency_unit: Option<FrequencyUnit>,
     pub metadata: serde_json::Value,
     pub executed_at: Option<DateTime<FixedOffset>>,
     pub error_message: Option<String>,
 }
 
+#[derive(Serialize, Deserialize, Debug, ToSql, FromSql)]
+#[postgres(name = "frequency_unit")]
+pub enum FrequencyUnit {
+    #[postgres(name = "days")]
+    Days,
+    #[postgres(name = "hours")]
+    Hours,
+    #[postgres(name = "minutes")]
+    Minutes,
+    #[postgres(name = "seconds")]
+    Seconds,
+}
+
 pub async fn insert_job(
     db: &DbClient, 
     name: &String,
     expected_time: &DateTime<FixedOffset>,
+    frequency: &Option<i32>,
+    frequency_unit: &Option<FrequencyUnit>,
     metadata: &serde_json::Value
 ) -> Result<()> {
     tracing::trace!("insert_job(name={})", name);
 
     db.execute(
-        "INSERT INTO jobs (name, expected_time, metadata) VALUES ($1, $2, $3) 
+        "INSERT INTO jobs (name, expected_time, frequency, frequency_unit, metadata) VALUES ($1, $2, $3, $4, $5) 
             ON CONFLICT (name, expected_time) DO UPDATE SET metadata = EXCLUDED.metadata",
-        &[&name, &expected_time, &metadata],
+        &[&name, &expected_time, &frequency, &frequency_unit, &metadata],
     )
     .await
     .context("Inserting job")?;
@@ -91,19 +109,32 @@ pub async fn get_jobs_to_execute(db: &DbClient) -> Result<Vec<Job>>  {
         let id: Uuid = job.get(0);
         let name: String = job.get(1);
         let expected_time: DateTime<FixedOffset> = job.get(2);
-        let metadata: serde_json::Value = job.get(3);
-        let executed_at: Option<DateTime<FixedOffset>> = job.get(4);
-        let error_message: Option<String> = job.get(5);
+        let frequency: Option<i32> = job.get(3);
+        let frequency_unit: Option<FrequencyUnit> = job.get(4);
+        let metadata: serde_json::Value = job.get(5);
+        let executed_at: Option<DateTime<FixedOffset>> = job.get(6);
+        let error_message: Option<String> = job.get(7);
 
         data.push(Job {
             id,
             name,
             expected_time,
-            metadata: metadata,
-            executed_at: executed_at,
+            frequency,
+            frequency_unit,
+            metadata,
+            executed_at,
             error_message
         });
     }
 
     Ok(data)
 }
+
+pub fn get_duration_from_cron(cron_period: i32, cron_unit: &FrequencyUnit) -> Duration {
+    match cron_unit {
+        FrequencyUnit::Days => Duration::days(cron_period as i64),
+        FrequencyUnit::Hours => Duration::hours(cron_period as i64),
+        FrequencyUnit::Minutes => Duration::minutes(cron_period as i64),
+        FrequencyUnit::Seconds => Duration::seconds(cron_period as i64),
+    }
+}