Rename expected_time to scheduled_at

Author: Mauricio Cassola

src/db.rs

@@ -184,39 +184,34 @@ pub async fn run_migrations(client: &DbClient) -> anyhow::Result<()> {
 
 pub async fn run_scheduled_jobs(db: &DbClient) -> anyhow::Result<()> {
     let jobs = get_jobs_to_execute(&db).await.unwrap();
-    println!("jobs to execute: {:#?}", jobs);
     tracing::trace!("jobs to execute: {:#?}", jobs);
 
     for job in jobs.iter() {
         update_job_executed_at(&db, &job.id).await?;
 
+        if let Some(frequency) = job.frequency {
+            let duration = get_duration_from_cron(frequency, job.frequency_unit.as_ref().unwrap());
+            let new_scheduled_at = job.scheduled_at.checked_add_signed(duration).unwrap();
+
+            insert_job(
+                &db,
+                &job.name,
+                &new_scheduled_at,
+                &Some(frequency),
+                &job.frequency_unit,
+                &job.metadata,
+            )
+            .await?;
+            tracing::trace!("job succesfully reinserted (name={})", job.name);
+        }
+
         match handle_job(&job.name, &job.metadata).await {
             Ok(_) => {
-                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) => {
-                println!("job failed on execution (id={:?}, error={:?})", job.id, e);
                 tracing::trace!("job failed on execution (id={:?}, error={:?})", job.id, e);
 
                 update_job_error_message(&db, &job.id, &e.to_string()).await?;
@@ -271,7 +266,7 @@ 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,
+    scheduled_at TIMESTAMP WITH TIME ZONE NOT NULL,
     frequency INTEGER,
     frequency_unit frequency_unit,
     metadata JSONB,
@@ -280,9 +275,9 @@ CREATE TABLE jobs (
 );
 ",
     "
-CREATE UNIQUE INDEX jobs_name_expected_time_unique_index 
+CREATE UNIQUE INDEX jobs_name_scheduled_at_unique_index 
     ON jobs (
-        name, expected_time
+        name, scheduled_at
     );
 ",
 ];

src/db/jobs.rs

@@ -10,7 +10,7 @@ use uuid::Uuid;
 pub struct Job {
     pub id: Uuid,
     pub name: String,
-    pub expected_time: DateTime<FixedOffset>,
+    pub scheduled_at: DateTime<FixedOffset>,
     pub frequency: Option<i32>,
     pub frequency_unit: Option<FrequencyUnit>,
     pub metadata: serde_json::Value,
@@ -34,17 +34,17 @@ pub enum FrequencyUnit {
 pub async fn insert_job(
     db: &DbClient,
     name: &String,
-    expected_time: &DateTime<FixedOffset>,
+    scheduled_at: &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, frequency, frequency_unit, metadata) VALUES ($1, $2, $3, $4, $5) 
-            ON CONFLICT (name, expected_time) DO UPDATE SET metadata = EXCLUDED.metadata",
-        &[&name, &expected_time, &frequency, &frequency_unit, &metadata],
+        "INSERT INTO jobs (name, scheduled_at, frequency, frequency_unit, metadata) VALUES ($1, $2, $3, $4, $5) 
+            ON CONFLICT (name, scheduled_at) DO UPDATE SET metadata = EXCLUDED.metadata",
+        &[&name, &scheduled_at, &frequency, &frequency_unit, &metadata],
     )
     .await
     .context("Inserting job")?;
@@ -86,13 +86,13 @@ pub async fn update_job_executed_at(db: &DbClient, id: &Uuid) -> Result<()> {
 }
 
 // Selects all jobs with:
-//  - expected_time in the past
+//  - scheduled_at in the past
 //  - error_message is null or executed_at is at least 60 minutes ago (intended to make repeat executions rare enough)
 pub async fn get_jobs_to_execute(db: &DbClient) -> Result<Vec<Job>> {
     let jobs = db
         .query(
             "
-        SELECT * FROM jobs WHERE expected_time <= now() AND (error_message IS NULL OR executed_at <= now() - INTERVAL '60 minutes')",
+        SELECT * FROM jobs WHERE scheduled_at <= now() AND (error_message IS NULL OR executed_at <= now() - INTERVAL '60 minutes')",
             &[],
         )
         .await
@@ -102,7 +102,7 @@ pub async fn get_jobs_to_execute(db: &DbClient) -> Result<Vec<Job>> {
     for job in jobs {
         let id: Uuid = job.get(0);
         let name: String = job.get(1);
-        let expected_time: DateTime<FixedOffset> = job.get(2);
+        let scheduled_at: DateTime<FixedOffset> = job.get(2);
         let frequency: Option<i32> = job.get(3);
         let frequency_unit: Option<FrequencyUnit> = job.get(4);
         let metadata: serde_json::Value = job.get(5);
@@ -112,7 +112,7 @@ pub async fn get_jobs_to_execute(db: &DbClient) -> Result<Vec<Job>> {
         data.push(Job {
             id,
             name,
-            expected_time,
+            scheduled_at,
             frequency,
             frequency_unit,
             metadata,

src/handlers/jobs.rs

@@ -20,7 +20,7 @@
 //
 //    Job {
 //      name: "send_zulip_message",
-//      expected_time: "2022-09-30T11:30:00+10:00",
+//      scheduled_at: "2022-09-30T11:30:00+10:00",
 //      frequency: Some(7),
 //      frequency_unit: Some(FrequencyUnit::Days),
 //      metadata: metadata
@@ -35,10 +35,6 @@ pub async fn handle_job(name: &String, metadata: &serde_json::Value) -> anyhow::
 }
 
 fn default(name: &String, metadata: &serde_json::Value) -> anyhow::Result<()> {
-    println!(
-        "handle_job fell into default case: (name={:?}, metadata={:?})",
-        name, metadata
-    );
     tracing::trace!(
         "handle_job fell into default case: (name={:?}, metadata={:?})",
         name,