execute a task

Cargo.toml

@@ -15,3 +15,5 @@ dotenv = "0.15.0"
 uuid = { version = "0.8", features = ["v4"] }
 chrono = "0.4"
 serde_json = "1.0"
+typetag = "0.1"
+serde = { version = "1.0", features = ["derive"] }

src/executor.rs

@@ -0,0 +1,125 @@
+use crate::postgres::Postgres;
+use crate::postgres::Task;
+
+struct Executor {}
+
+#[derive(Debug)]
+pub struct Error {
+    description: String,
+}
+
+#[typetag::serde(tag = "type")]
+pub trait Runnable {
+    fn run(&self) -> Result<(), Error>;
+}
+
+impl Executor {
+    pub fn run(storage: &Postgres, task: &Task) {
+        let actual_task: Box<dyn Runnable> = serde_json::from_value(task.metadata.clone()).unwrap();
+
+        match actual_task.run() {
+            Ok(()) => storage.finish_task(task).unwrap(),
+            Err(error) => storage.fail_task(task, error.description).unwrap(),
+        };
+    }
+}
+
+#[cfg(test)]
+mod executor_tests {
+    use super::Error;
+    use super::Executor;
+    use super::Runnable;
+    use crate::postgres::NewTask;
+    use crate::postgres::Postgres;
+    use crate::schema::FangTaskState;
+    use diesel::connection::Connection;
+    use serde::{Deserialize, Serialize};
+
+    #[derive(Serialize, Deserialize)]
+    struct Job {
+        pub number: u16,
+    }
+
+    #[typetag::serde]
+    impl Runnable for Job {
+        fn run(&self) -> Result<(), Error> {
+            println!("the number is {}", self.number);
+
+            Ok(())
+        }
+    }
+
+    #[derive(Serialize, Deserialize)]
+    struct FailedJob {
+        pub number: u16,
+    }
+
+    #[typetag::serde]
+    impl Runnable for FailedJob {
+        fn run(&self) -> Result<(), Error> {
+            let message = format!("the number is {}", self.number);
+
+            Err(Error {
+                description: message,
+            })
+        }
+    }
+
+    pub fn serialize(job: &dyn Runnable) -> serde_json::Value {
+        serde_json::to_value(job).unwrap()
+    }
+
+    #[test]
+    fn executes_and_finishes_task() {
+        let job = Job { number: 10 };
+
+        let postgres = Postgres::new(None);
+
+        let new_task = NewTask {
+            metadata: serialize(&job),
+        };
+
+        postgres.connection.test_transaction::<(), Error, _>(|| {
+            let task = postgres.insert(&new_task).unwrap();
+
+            assert_eq!(FangTaskState::New, task.state);
+
+            Executor::run(&postgres, &task);
+
+            let found_task = postgres.find_task_by_id(task.id).unwrap();
+
+            assert_eq!(FangTaskState::Finished, found_task.state);
+
+            Ok(())
+        });
+    }
+
+    #[test]
+    fn saves_error_for_failed_task() {
+        let job = FailedJob { number: 10 };
+
+        let postgres = Postgres::new(None);
+
+        let new_task = NewTask {
+            metadata: serialize(&job),
+        };
+
+        postgres.connection.test_transaction::<(), Error, _>(|| {
+            let task = postgres.insert(&new_task).unwrap();
+
+            assert_eq!(FangTaskState::New, task.state);
+
+            Executor::run(&postgres, &task);
+
+            let found_task = postgres.find_task_by_id(task.id).unwrap();
+
+            assert_eq!(FangTaskState::Failed, found_task.state);
+            assert_eq!(
+                "the number is 10".to_string(),
+                found_task.error_message.unwrap()
+            );
+
+            Ok(())
+        });
+    }
+}

src/lib.rs

@@ -15,6 +15,7 @@ extern crate diesel;
 //     fn save() ->
 // }
 
+pub mod executor;
 pub mod postgres;
 mod schema;
 

src/postgres.rs

@@ -70,6 +70,39 @@ impl Postgres {
             _ => None,
         }
     }
+
+    pub fn find_task_by_id(&self, id: Uuid) -> Option<Task> {
+        match fang_tasks::table
+            .filter(fang_tasks::id.eq(id))
+            .first::<Task>(&self.connection)
+        {
+            Ok(record) => Some(record),
+            _ => None,
+        }
+    }
+
+    pub fn finish_task(&self, task: &Task) -> Result<Task, Error> {
+        diesel::update(task)
+            .set((
+                fang_tasks::state.eq(FangTaskState::Finished),
+                fang_tasks::updated_at.eq(Self::current_time()),
+            ))
+            .get_result::<Task>(&self.connection)
+    }
+
+    pub fn fail_task(&self, task: &Task, error: String) -> Result<Task, Error> {
+        diesel::update(task)
+            .set((
+                fang_tasks::state.eq(FangTaskState::Failed),
+                fang_tasks::error_message.eq(error),
+                fang_tasks::updated_at.eq(Self::current_time()),
+            ))
+            .get_result::<Task>(&self.connection)
+    }
+
+    pub fn current_time() -> DateTime<Utc> {
+        Utc::now()
+    }
 }
 
 #[cfg(test)]
@@ -121,7 +154,39 @@ mod postgres_tests {
         });
     }
 
-    // this test ignored because it commits data to the db
+    #[test]
+    fn finish_task_updates_state_field() {
+        let postgres = Postgres::new(None);
+
+        postgres.connection.test_transaction::<(), Error, _>(|| {
+            let task = insert_new_job(&postgres.connection);
+
+            let updated_task = postgres.finish_task(&task).unwrap();
+
+            assert_eq!(FangTaskState::Finished, updated_task.state);
+
+            Ok(())
+        });
+    }
+
+    #[test]
+    fn fail_task_updates_state_field_and_sets_error_message() {
+        let postgres = Postgres::new(None);
+
+        postgres.connection.test_transaction::<(), Error, _>(|| {
+            let task = insert_new_job(&postgres.connection);
+            let error = "Failed".to_string();
+
+            let updated_task = postgres.fail_task(&task, error.clone()).unwrap();
+
+            assert_eq!(FangTaskState::Failed, updated_task.state);
+            assert_eq!(error, updated_task.error_message.unwrap());
+
+            Ok(())
+        });
+    }
+
+    // this test is ignored because it commits data to the db
     #[test]
     #[ignore]
     fn fetch_task_locks_the_record() {
@@ -178,4 +243,11 @@ mod postgres_tests {
             .get_result::<Task>(connection)
             .unwrap()
     }
+
+    fn insert_new_job(connection: &PgConnection) -> Task {
+        diesel::insert_into(fang_tasks::table)
+            .values(&vec![(fang_tasks::metadata.eq(serde_json::json!(true)),)])
+            .get_result::<Task>(connection)
+            .unwrap()
+    }
 }