ref, feat(scheduler) global project refactor, add scheduler

backend/api_server/src/blog/api/get_page_count.rs

@@ -0,0 +1,26 @@
+use constants::blog::PAGE_SIZE;
+
+use axum::extract::State;
+use axum::Json;
+use serde_json::{json, Value};
+use sqlx::{query, PgPool, Row};
+use tracing::error;
+
+pub async fn get_page_count(
+    State(db_connection): State<PgPool>,
+) -> Result<Json<Value>, Json<Value>> {
+    match query("select cast(ceil(count(*) / $1) as integer) as res from post")
+        .bind(PAGE_SIZE as f32)
+        .fetch_one(&db_connection)
+        .await
+    {
+        Ok(res) => {
+            let res: i32 = res.get("res");
+            Ok(Json::from(json!({"res": res})))
+        }
+        Err(e) => {
+            error!("Failed to get blog page count: {}", e.to_string());
+            Err(Json::from(json!({"error": e.to_string()})))
+        }
+    }
+}

backend/api_server/src/blog/api/get_post.rs

@@ -0,0 +1,153 @@
+use axum::{
+    extract::{Path, State},
+    http::StatusCode,
+    Json,
+};
+use serde_json::{json, Value};
+use sqlx::{query, Error, PgPool, Row};
+use std::str::FromStr;
+use tracing::error;
+use uuid::Uuid;
+
+pub async fn get_post(
+    State(db_connection): State<PgPool>,
+    Path(post_id): Path<String>,
+) -> Result<Json<Value>, (StatusCode, Json<Value>)> {
+    let post_id = match Uuid::from_str(post_id.as_str()) {
+        Ok(id) => id,
+        Err(e) => {
+            error!(
+                "Failed to parse UUID <{}>: {}",
+                post_id.as_str(),
+                e.to_string()
+            );
+            return Err((
+                StatusCode::BAD_REQUEST,
+                Json::from(json!(("error", e.to_string()))),
+            ));
+        }
+    };
+
+    let post = match query(
+        r#"
+SELECT jsonb_strip_nulls(to_jsonb(t.*)) - 'post_id' AS post
+FROM post t
+WHERE post_id = $1;
+"#,
+    )
+    .bind(post_id)
+    .fetch_one(&db_connection)
+    .await
+    {
+        Ok(value) => value.get::<Value, _>("post"),
+        Err(e) => {
+            error!(
+                "Database error on Reading `base post information` from table `post`: <{}>: {}",
+                post_id.to_string(),
+                e.to_string()
+            );
+            return match e {
+                Error::RowNotFound => Err((
+                    StatusCode::BAD_REQUEST,
+                    Json::from(json!({"error": e.to_string()})),
+                )),
+                _ => Err((
+                    StatusCode::INTERNAL_SERVER_ERROR,
+                    Json::from(json!({"error": e.to_string()})),
+                )),
+            };
+        }
+    };
+
+    let tags: Vec<_> = match query("SELECT tag FROM Tag WHERE post_id = $1;")
+        .bind(post_id)
+        .fetch_all(&db_connection)
+        .await
+    {
+        Ok(tags) => tags.iter().map(|tag| tag.get::<String, _>("tag")).collect(),
+        Err(e) => {
+            error!("Database query error :{}", e.to_string());
+            return Err((
+                StatusCode::INTERNAL_SERVER_ERROR,
+                Json::from(json!({"error": e.to_string()})),
+            ));
+        }
+    };
+
+    let meta = match query(
+        r#"
+SELECT jsonb_strip_nulls(to_jsonb(t.*)) - 'post_id' AS meta
+FROM meta t
+WHERE post_id = $1;
+        "#,
+    )
+    .bind(post_id)
+    .fetch_one(&db_connection)
+    .await
+    {
+        Ok(value) => value.get::<Value, _>("meta"),
+        Err(e) => {
+            error!(
+                "Database error on Reading `{}` from table `{}`: <{}>: {}",
+                "post meta data",
+                "meta",
+                post_id.to_string(),
+                e.to_string()
+            );
+            return Err((
+                StatusCode::INTERNAL_SERVER_ERROR,
+                Json::from(json!({"error": e.to_string()})),
+            ));
+        }
+    };
+
+    let adjacent = match query(r"
+WITH target_post AS (SELECT t.date FROM meta t WHERE post_id = $1),
+     prev AS (SELECT post_id, t.title, t.sub_title, t.header_img
+              FROM meta t
+              WHERE t.date < (SELECT date FROM target_post)
+              ORDER BY t.date DESC
+              LIMIT 1),
+     next AS (SELECT post_id, t.title, t.sub_title, t.header_img
+              FROM meta t
+              WHERE t.date > (SELECT date FROM target_post)
+              ORDER BY t.date
+              LIMIT 1)
+SELECT json_strip_nulls(json_build_object(
+        'prev',
+        (SELECT json_build_object('id', post_id, 'title', title, 'sub_title', sub_title, 'header_img', header_img)
+         FROM prev),
+        'next',
+        (SELECT json_build_object('id', post_id, 'title', title, 'sub_title', sub_title, 'header_img', header_img)
+         FROM next)
+                        )) AS result;
+"
+    )
+        .bind(post_id)
+        .fetch_one(&db_connection)
+        .await
+    {
+        Ok(res) => {res.get::<Value, _>("result")}
+        Err(e) => {
+            error!(
+                "Database error on Reading `{}` from table `{}`: <{}>: {}",
+                "adjoint posts data",
+                "meta",
+                post_id.to_string(),
+                e.to_string()
+            );
+            return Err((
+                StatusCode::INTERNAL_SERVER_ERROR,
+                Json::from(json!({"error": e.to_string()})),
+            ));
+        }
+
+    };
+
+    let mut res = post.as_object().unwrap().to_owned();
+    res.insert("tags".to_string(), tags.into());
+    res.insert("meta".to_string(), meta);
+    res.insert("adj".to_string(), adjacent);
+
+    Ok(Json::from(json!(res)))
+}

backend/api_server/src/blog/api/get_post_list.rs

@@ -0,0 +1,88 @@
+use constants::blog::PAGE_SIZE;
+
+use axum::{extract::Query, extract::State, http::StatusCode, Json};
+use chrono::{DateTime, Utc};
+use futures::future::join_all;
+use serde::Deserialize;
+use serde_json::{json, Map, Value};
+use sqlx::{postgres::PgRow, query, PgPool, Row};
+use tracing::error;
+use uuid::Uuid;
+
+#[derive(Deserialize)]
+pub struct Params {
+    page: Option<u32>,
+}
+
+pub async fn get_post_list(
+    State(db_connection): State<PgPool>,
+    Query(params): Query<Params>,
+) -> Result<Json<Value>, (StatusCode, Json<Value>)> {
+    let search_param_page = params.page.unwrap_or(1);
+    if search_param_page < 1 {
+        return Err((
+            StatusCode::BAD_REQUEST,
+            Json::from(json!({"error": "`page` should be greater than 0"})),
+        ));
+    }
+
+    let offset = PAGE_SIZE * (search_param_page - 1);
+
+    let post_list = sqlx::query("SELECT * FROM post ORDER BY first_update DESC LIMIT $1 OFFSET $2")
+        .bind(PAGE_SIZE as i32)
+        .bind(offset as i32)
+        .fetch_all(&db_connection)
+        .await;
+    let post_list = match post_list {
+        Ok(post_list) => post_list,
+        Err(e) => {
+            error!(
+                "Database query error on fetching post list: {}",
+                e.to_string()
+            );
+            return Err((
+                StatusCode::INTERNAL_SERVER_ERROR,
+                Json::from(json!({"error": e.to_string()})),
+            ));
+        }
+    };
+
+    let mut res = Map::new();
+    let posts: Vec<_> = post_list
+        .into_iter()
+        .map(|post| analyze_post(&db_connection, post))
+        .collect();
+
+    let posts = join_all(posts).await;
+
+    res.insert("posts".to_string(), Value::from(posts));
+    Ok(Json::from(Value::from(res)))
+}
+
+async fn analyze_post(db_connection: &PgPool, post: PgRow) -> Value {
+    let post_id: Uuid = post.get("post_id");
+    let tags = fetch_post_tags(db_connection, post_id).await;
+
+    json!({
+        "post_id": post_id.to_string(),
+        "title": post.get::<String, _>("title"),
+        "category": post.get::<String, _>("category"),
+        "header_img": post.get::<String, _>("header_img"),
+        "sub_title": post.get::<String, _>("sub_title"),
+        "summary": post.get::<String, _>("summary"),
+        "last_update": post.get::<DateTime<Utc>, _>("last_update"),
+        "first_update": post.get::<DateTime<Utc>, _>("first_update"),
+        "tags": tags,
+    })
+}
+
+async fn fetch_post_tags(db_connection: &PgPool, post_id: Uuid) -> Vec<String> {
+    match query("SELECT tag FROM Tag WHERE post_id = $1")
+        .bind(post_id)
+        .fetch_all(db_connection)
+        .await
+    {
+        Ok(res) => res.iter().map(|tag| tag.get::<String, _>("tag")).collect(),
+        Err(_) => vec![],
+    }
+}

backend/api_server/src/blog/api/mod.rs

@@ -0,0 +1,4 @@
+pub mod get_page_count;
+pub mod get_post;
+pub mod get_post_list;
+pub mod refresh;

backend/api_server/src/blog/api/refresh.rs

@@ -0,0 +1,190 @@
+use std::path::Path;
+use std::str::FromStr;
+
+use axum::extract::{Query, State};
+use axum::http::StatusCode;
+use axum::Json;
+use serde::Deserialize;
+use serde_json::{Map, Value};
+use sqlx::PgPool;
+
+// use crate::api::post::errors::refresh::PostIdxError;
+// use crate::api::post::utils::{build_post_idx, post};
+// use crate::model::blog::{POST_BASE_MODEL, POST_HASH_MODEL, POST_META_MODEL, TAG_MODEL};
+
+#[derive(Deserialize)]
+pub struct Params {
+    /// refresh level
+    level: Option<String>,
+}
+
+pub async fn refresh(
+    State(db_connection): State<PgPool>,
+    Query(params): Query<Params>,
+) -> Result<StatusCode, (StatusCode, Json<Value>)> {
+    // let is_force_refresh = params.level.unwrap_or("".to_string()) == "force";
+    //
+    // // search post
+    // let data_repo_dir = match env::var("GIT_WORK_DIR") {
+    //     Ok(value) => value,
+    //     Err(_) => {
+    //         return Err((
+    //             StatusCode::INTERNAL_SERVER_ERROR,
+    //             Json::from(json!((
+    //                 "error",
+    //                 "error at parsing environment variable: GIT_WORK_DIR"
+    //             ))),
+    //         ));
+    //     }
+    // };
+    // let post_dir = Path::new(&data_repo_dir).join("posts");
+    //
+    // if is_force_refresh {
+    //     if let Err(e) = query("DROP TABLE IF EXISTS hash, meta, post, tag;")
+    //         .execute(&db_connection)
+    //         .await
+    //     {
+    //         error!(
+    //             "DB query error when deleting table: `hash`, `meta`, `post`, `tag`: {}",
+    //             e.to_string()
+    //         );
+    //         return Err((
+    //             StatusCode::INTERNAL_SERVER_ERROR,
+    //             Json::from(json!(("error", e.to_string()))),
+    //         ));
+    //     }
+    // }
+    //
+    // // generate post idx
+    // let post_path = search_md(&post_dir);
+    //
+    // let post_list_with_hash: Vec<(Map<_, _>, Vec<_>)> = post_path
+    //     .into_iter()
+    //     .filter_map(|post_path| match post::from_path(&post_path) {
+    //         Ok(post) => match fs::read(&post_path) {
+    //             Ok(bytes) => {
+    //                 let mut hasher = Sha256::new();
+    //                 hasher.update(bytes);
+    //                 let hash = hasher.finalize().to_vec();
+    //                 Some((post, hash))
+    //             }
+    //             Err(e) => {
+    //                 error!(
+    //                     "Failed to read file while hashing the post <{}>, skipping: {}",
+    //                     post_path,
+    //                     e.to_string()
+    //                 );
+    //                 None
+    //             }
+    //         },
+    //         Err(e) => {
+    //             error!(
+    //                 "Failed to parse post <{}>, skipping: {}",
+    //                 post_path,
+    //                 e.to_string()
+    //             );
+    //             None
+    //         }
+    //     })
+    //     .collect();
+    //
+    // // db create
+    // query(POST_BASE_MODEL)
+    //     .execute(&db_connection)
+    //     .await
+    //     .unwrap();
+    // query(POST_META_MODEL)
+    //     .execute(&db_connection)
+    //     .await
+    //     .unwrap();
+    // query(TAG_MODEL).execute(&db_connection).await.unwrap();
+    // query(POST_HASH_MODEL)
+    //     .execute(&db_connection)
+    //     .await
+    //     .unwrap();
+    //
+    // let res: Vec<_> = stream::iter(post_list_with_hash)
+    //     .map(|(post, hash)| {
+    //         let db_clone = db_connection.clone();
+    //         async move { build_post_idx(&db_clone, post, hash).await }
+    //     })
+    //     .buffer_unordered(usize::MAX)
+    //     .collect()
+    //     .await;
+    //
+    // let errs: Vec<_> = res
+    //     .into_iter()
+    //     .filter_map(|res| res.err())
+    //     .map(|e| e.to_string())
+    //     .collect();
+    //
+    // if !errs.is_empty() {
+    //     Err((
+    //         StatusCode::INTERNAL_SERVER_ERROR,
+    //         Json::from(json!({"error": errs})),
+    //     ))
+    // } else {
+    //     Ok(StatusCode::OK)
+    // }
+    todo!()
+}
+
+/// Search a dir, return a list of markdown files
+///
+/// # Arguments
+/// * `entry`: dir path
+///
+/// # Returns
+/// Vec<String, Global>
+/// * `String`: path of a markdown file
+///
+fn search_md(entry: &Path) -> Vec<String> {
+    if entry.ends_with(".git") {
+        return vec![];
+    }
+    if entry.is_dir() {
+        if let Ok(files) = entry.read_dir() {
+            files
+                .flat_map(|file| {
+                    let file_path = file.unwrap().path();
+                    search_md(file_path.as_path())
+                })
+                .collect()
+        } else {
+            vec![]
+        }
+    } else if entry.extension().is_some_and(|ext| ext == "md") {
+        vec![entry.to_str().unwrap().to_string()]
+    } else {
+        vec![]
+    }
+}
+
+async fn build_post_idx(
+    db_connection: &PgPool,
+    post: Map<String, Value>,
+    hash: Vec<u8>,
+    // ) -> Result<(), PostIdxError> {
+) -> Result<(), ()> {
+    todo!()
+    // let meta = &post["meta"];
+    //
+    // // extract post id
+    // let post_id = match Uuid::from_str(post["post_id"].as_str().unwrap()) {
+    //     Ok(id) => id,
+    //     Err(e) => {
+    //         return Err(PostIdxError::InvalidUUID {
+    //             id: post["post_id"].as_str().unwrap().to_string(),
+    //             err_msg: e.to_string(),
+    //         })
+    //     }
+    // };
+    //
+    // if build_post_idx::hash(db_connection, post_id, hash).await? {
+    //     build_post_idx::base(db_connection, &post, post_id, meta).await?;
+    //     build_post_idx::tag(db_connection, post_id, meta).await?;
+    //     build_post_idx::meta(db_connection, post_id, meta).await?;
+    // }
+    //
+    // Ok(())
+}

backend/api_server/src/blog/mod.rs

@@ -0,0 +1,19 @@
+mod api;
+mod scheduler;
+
+use crate::blog::api::{
+    get_page_count::get_page_count, get_post::get_post, get_post_list::get_post_list,
+};
+
+use crate::blog::api::refresh::refresh;
+use axum::routing::{get, post};
+use axum::Router;
+use sqlx::{Pool, Postgres};
+
+pub fn get_router() -> Router<Pool<Postgres>> {
+    Router::new()
+        .route("/refresh_posts", post(refresh))
+        .route("/get_post_list", get(get_post_list))
+        .route("/get_page_count", get(get_page_count))
+        .route("/get_post/:post_id", get(get_post))
+}

backend/api_server/src/blog/scheduler/mod.rs

@@ -0,0 +1,184 @@
+use crate::blog::{
+    scheduler::scheduler_error::SchedulerError,
+    scheduler::scheduler_error::SchedulerError::IdAlreadyRunning,
+    scheduler::scheduler_error::SchedulerError::IdNotFound, scheduler::sleeper::SkippableSleeper,
+    scheduler::tasks::Task,
+};
+use std::sync::atomic::{AtomicBool, Ordering};
+use std::sync::RwLock;
+use std::{collections::HashMap, sync::Arc, thread};
+use uuid::Uuid;
+
+pub mod scheduler_error;
+mod sleeper;
+mod task_func;
+pub mod tasks;
+
+struct TaskStatus {
+    sleeper: Option<Arc<SkippableSleeper>>,
+    is_alive: AtomicBool,
+}
+
+#[derive(Default)]
+pub struct Scheduler {
+    tasks: Arc<RwLock<HashMap<Uuid, Arc<Box<dyn Task>>>>>,
+    running: Arc<RwLock<HashMap<Uuid, Arc<TaskStatus>>>>,
+    stopped: Arc<RwLock<Vec<Uuid>>>,
+}
+
+impl Scheduler {
+    /// Start all tasks in the scheduler
+    pub fn start(&self) -> Result<(), SchedulerError> {
+        for (k, _v) in self.tasks.read()?.iter() {
+            self.run_non_blocking(k)?
+        }
+        Ok(())
+    }
+
+    /// Terminate all tasks in the scheduler
+    pub fn halt() {}
+
+    fn run_task(
+        running: Arc<RwLock<HashMap<Uuid, Arc<TaskStatus>>>>,
+        stopped: Arc<RwLock<Vec<Uuid>>>,
+        id: Uuid,
+        task: Arc<Box<dyn Task>>,
+    ) -> Result<(), SchedulerError> {
+        if running.read()?.get(&id).is_some() {
+            return Err(IdAlreadyRunning);
+        } else {
+            running.write()?.insert(
+                id,
+                TaskStatus {
+                    sleeper: None,
+                    is_alive: AtomicBool::new(true),
+                }
+                .into(),
+            );
+        }
+        loop {
+            if let Some(next_execution) = task.get_next_execution() {
+                // guarantee is alive
+                let task_status = {
+                    let running_reading_guard = running.read()?;
+                    running_reading_guard.get(&id).cloned().clone()
+                };
+                if let Some(task_status) = task_status {
+                    if !&task_status.is_alive.load(Ordering::SeqCst) {
+                        {
+                            let mut running_lock_guard = running.write()?;
+                            running_lock_guard.remove(&id);
+                        }
+                        stopped.write()?.push(id);
+                        return Ok(());
+                    } else {
+                        let sleeper = Arc::from(SkippableSleeper::from(next_execution));
+                        {
+                            let mut running_lock_guard = running.write()?;
+                            running_lock_guard.insert(
+                                id,
+                                TaskStatus {
+                                    sleeper: Some(sleeper.clone()),
+                                    is_alive: AtomicBool::new(true),
+                                }
+                                .into(),
+                            );
+                        }
+                        sleeper.start();
+                        task.call();
+                    }
+                }
+            } else {
+                stopped.write()?.push(id);
+                break;
+            }
+        }
+        Ok(())
+    }
+
+    /// Run a task with `id`
+    pub fn run(&self, id: &Uuid) -> Result<(), SchedulerError> {
+        let id = id.to_owned();
+        if let Some(task) = self.tasks.read()?.get(&id) {
+            Self::run_task(self.running.clone(), self.stopped.clone(), id, task.clone())
+        } else {
+            Err(IdNotFound)
+        }
+    }
+
+    pub fn run_non_blocking(&self, id: &Uuid) -> Result<(), SchedulerError> {
+        let id = id.to_owned();
+        if let Some(task) = self.tasks.read()?.get(&id) {
+            let running_clone = self.running.clone();
+            let stopped_clone = self.stopped.clone();
+            let task_clone = task.clone();
+            thread::spawn(move || Self::run_task(running_clone, stopped_clone, id, task_clone));
+            Ok(())
+        } else {
+            Err(IdNotFound)
+        }
+    }
+
+    fn get_running(&self, id: &Uuid) -> Option<Arc<TaskStatus>> {
+        let ret = self.running.read().unwrap().get(id).cloned();
+        ret
+    }
+
+    /// Stop a task with `id`
+    pub fn stop(&self, id: &Uuid) -> Result<(), SchedulerError> {
+        match self.get_running(id) {
+            None => Err(IdNotFound),
+            Some(status) => {
+                status.is_alive.store(false, Ordering::SeqCst);
+                if let Some(s) = &status.sleeper {
+                    s.skip();
+                } else {
+                    return Err(IdAlreadyRunning);
+                }
+                Ok(())
+            }
+        }
+    }
+
+    pub fn now(&self, id: &Uuid) -> Result<(), SchedulerError> {
+        match self.get_running(id) {
+            None => Err(IdNotFound),
+            Some(status) => {
+                if let Some(s) = &status.sleeper {
+                    s.skip()
+                } else {
+                    return Err(IdAlreadyRunning);
+                }
+                Ok(())
+            }
+        }
+    }
+
+    pub fn insert(&self, task: Box<dyn Task>) -> Result<Uuid, SchedulerError> {
+        let id = Uuid::new_v4();
+        self.tasks.write()?.insert(id, Arc::new(task));
+        Ok(id)
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use crate::blog::scheduler::tasks::basic::BasicTask;
+    use crate::blog::scheduler::Scheduler;
+    use std::error::Error;
+    use std::thread;
+    use std::time::Duration;
+
+    #[test]
+    fn test_run() -> Result<(), Box<dyn Error>> {
+        let scheduler = Scheduler::default();
+        let task = BasicTask::new(|| println!("test print"), "* * * * * *")?;
+        let id = scheduler.insert(Box::new(task))?;
+        scheduler.run_non_blocking(&id)?;
+        thread::sleep(Duration::from_secs(4));
+        scheduler.stop(&id)?;
+        // test termination
+        thread::sleep(Duration::from_secs(5));
+        Ok(())
+    }
+}

backend/api_server/src/blog/scheduler/scheduler_error.rs

@@ -0,0 +1,44 @@
+use crate::blog::scheduler::scheduler_error::SchedulerError::MutexPoison;
+use std::error::Error;
+use std::fmt::{Debug, Display, Formatter};
+use std::sync::PoisonError;
+
+pub enum SchedulerError {
+    IdNotFound,
+    IdAlreadyRunning,
+    MutexPoison(String),
+}
+
+impl<T> From<PoisonError<T>> for SchedulerError {
+    fn from(value: PoisonError<T>) -> Self {
+        MutexPoison(value.to_string())
+    }
+}
+
+impl Debug for SchedulerError {
+    fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
+        write!(f, "{}", self)
+    }
+}
+
+impl Display for SchedulerError {
+    fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
+        write!(
+            f,
+            "{}",
+            match self {
+                SchedulerError::IdNotFound => {
+                    "SchedulerError::IdNotFound".to_string()
+                }
+                SchedulerError::IdAlreadyRunning => {
+                    "SchedulerError::IdAlreadyRunning".to_string()
+                }
+                MutexPoison(str) => {
+                    format!("SchedulerError::MutexPoison: {}", str)
+                }
+            }
+        )
+    }
+}
+
+impl Error for SchedulerError {}

backend/api_server/src/blog/scheduler/sleeper.rs

@@ -0,0 +1,77 @@
+use chrono::{DateTime, Utc};
+use std::sync::{Arc, Condvar, Mutex};
+use std::thread;
+use std::time::Duration;
+
+#[derive(Clone, Debug, Default)]
+pub struct SkippableSleeper {
+    pub duration: Duration,
+    state: Arc<(Mutex<bool>, Condvar)>,
+}
+
+impl SkippableSleeper {
+    pub fn new(duration: Duration) -> Self {
+        SkippableSleeper {
+            duration,
+            state: Arc::new((Mutex::new(false), Condvar::new())),
+        }
+    }
+
+    // Blocks until the duration has passed or skip() is called
+    pub fn start(&self) {
+        let (lock, cvar) = &*self.state;
+        let mut received = lock.lock().unwrap();
+        let state_clone = Arc::clone(&self.state);
+        let duration_clone = self.duration;
+        thread::spawn(move || {
+            thread::sleep(duration_clone);
+            let (lock, cvar) = &*state_clone;
+            let mut received = lock.lock().unwrap();
+            *received = true;
+            cvar.notify_one();
+        });
+        while !*received {
+            received = cvar.wait(received).unwrap();
+        }
+    }
+
+    // Skip the blocking delay
+    pub fn skip(&self) {
+        let (lock, cvar) = &*self.state;
+        let mut received = lock.lock().unwrap();
+        *received = true;
+        cvar.notify_one();
+    }
+}
+
+impl From<DateTime<Utc>> for SkippableSleeper {
+    fn from(value: DateTime<Utc>) -> Self {
+        let from_utc_timestamp = value.timestamp() as u64;
+        let curr_utc_timestamp = Utc::now().timestamp() as u64;
+        SkippableSleeper::new(Duration::from_secs(from_utc_timestamp - curr_utc_timestamp))
+    }
+}
+
+#[cfg(test)]
+mod test {
+    use crate::blog::scheduler::sleeper::SkippableSleeper;
+    use std::thread;
+    use std::time::Duration;
+
+    #[test]
+    fn usage() {
+        let sleeper = SkippableSleeper::new(Duration::new(10, 0));
+        thread::spawn({
+            let sleeper = sleeper.clone();
+            move || {
+                println!("Sleeping...");
+                sleeper.start();
+                println!("Woke up!");
+            }
+        });
+        println!("Start Sleeping...");
+        thread::sleep(Duration::from_secs(4));
+        println!("Skipping sleep...");
+        sleeper.skip();
+    }
+}

backend/api_server/src/blog/scheduler/task_func.rs

@@ -0,0 +1,3 @@
+pub trait TaskFunc: Fn() + Send + Sync + 'static {}
+
+impl<T> TaskFunc for T where T: Fn() + Send + Sync + Clone + 'static {}

backend/api_server/src/blog/scheduler/tasks/basic.rs

@@ -0,0 +1,30 @@
+use crate::blog::scheduler::task_func::TaskFunc;
+use crate::blog::scheduler::tasks::Task;
+use chrono::{DateTime, Utc};
+use cron::Schedule;
+use std::error::Error;
+use std::str::FromStr;
+
+pub struct BasicTask {
+    schedule: Schedule,
+    func: Box<dyn TaskFunc>,
+}
+
+impl BasicTask {
+    pub fn new(func: fn(), cron_expr: &str) -> Result<Self, Box<dyn Error>> {
+        Ok(BasicTask {
+            schedule: Schedule::from_str(cron_expr)?,
+            func: Box::new(func),
+        })
+    }
+}
+
+impl Task for BasicTask {
+    fn get_next_execution(&self) -> Option<DateTime<Utc>> {
+        self.schedule.upcoming(Utc).next()
+    }
+
+    fn call(&self) {
+        (self.func)()
+    }
+}

backend/api_server/src/blog/scheduler/tasks/mod.rs

@@ -0,0 +1,9 @@
+use crate::blog::scheduler::task_func::TaskFunc;
+use chrono::{DateTime, Utc};
+
+pub mod basic;
+
+pub trait Task: Send + Sync {
+    fn get_next_execution(&self) -> Option<DateTime<Utc>>;
+    fn call(&self);
+}

backend/api_server/src/init.rs

@@ -1,5 +1,7 @@
-use crate::logger;
 use std::env;
+
+use crate::logger;
+
 use tracing_appender::non_blocking::WorkerGuard;
 
 pub fn logger_init() -> (WorkerGuard, WorkerGuard) {

backend/api_server/src/logger.rs

@@ -1,16 +1,14 @@
-use chrono::Local;
-use dirs::home_dir;
 use std::path::Path;
 use std::{env, io};
+
+use chrono::Local;
+use dirs::home_dir;
 use tracing::info;
-use tracing_appender::non_blocking::WorkerGuard;
-use tracing_appender::{non_blocking, rolling};
-use tracing_subscriber::filter::LevelFilter;
-use tracing_subscriber::fmt::format::Writer;
-use tracing_subscriber::fmt::time::FormatTime;
-use tracing_subscriber::fmt::MakeWriter;
-use tracing_subscriber::layer::SubscriberExt;
-use tracing_subscriber::{fmt, Layer, Registry};
+use tracing_appender::{non_blocking, non_blocking::WorkerGuard, rolling};
+use tracing_subscriber::{
+    filter::LevelFilter, fmt, fmt::format::Writer, fmt::time::FormatTime, fmt::MakeWriter,
+    layer::SubscriberExt, Layer, Registry,
+};
 
 struct LocalTimer;
 
@@ -95,10 +93,10 @@ pub fn init() -> (WorkerGuard, WorkerGuard) {
         );
 
     tracing::subscriber::set_global_default(subscribers)
-        .expect("[fatal] Failed to setup logging system");
+        .expect("[fatal] failed to setup logging system");
 
-    info!("Module Started: Logger");
-    info!("Logging to file {}", log_path);
+    info!("module started: logger");
+    info!("logging to dir: {}", log_path);
 
     (info_guard, error_guard)
 }

backend/api_server/src/main.rs

@@ -1,7 +1,12 @@
+mod blog;
 mod init;
 mod logger;
+mod server;
 
-fn main() {
+#[tokio::main]
+async fn main() {
     init::load_env();
     let _guards = init::logger_init();
+
+    server::start().await;
 }

backend/api_server/src/server.rs

@@ -0,0 +1,65 @@
+use std::env;
+use std::num::ParseIntError;
+use std::process::exit;
+use std::str::FromStr;
+
+use crate::blog;
+use axum::Router;
+use sqlx::{
+    postgres::{PgConnectOptions, PgPoolOptions},
+    ConnectOptions,
+};
+use tracing::{error, info, warn};
+
+pub async fn start() {
+    let port: u32 = match env::var("BACKEND_PORT") {
+        Ok(value) => value.parse().unwrap_or_else(|e: ParseIntError| {
+            warn!("failed to parse BACKEND_PORT: {}", e.to_string());
+            3001
+        }),
+        Err(_) => 3001,
+    };
+
+    let db_connect_option = PgConnectOptions::from_str(&parse_db_uri())
+        .unwrap()
+        .disable_statement_logging();
+
+    let db_pool = match PgPoolOptions::new().connect_with(db_connect_option).await {
+        Ok(pool) => {
+            info!("database connected");
+            pool
+        }
+        Err(e) => {
+            error!("failed to load database: {}", e.to_string());
+            exit(1);
+        }
+    };
+
+    let app = Router::new()
+        .nest("blog", blog::get_router())
+        .with_state(db_pool);
+
+    let listener = tokio::net::TcpListener::bind(format!("0.0.0.0:{}", port))
+        .await
+        .unwrap_or_else(|e| {
+            error!("failed to build TCP listener: {}", e.to_string());
+            exit(1);
+        });
+
+    axum::serve(listener, app).await.unwrap_or_else(|e| {
+        error!("failed to start axum: {}", e.to_string());
+        exit(1);
+    });
+}
+
+fn parse_db_uri() -> String {
+    let postgres_user = env::var("POSTGRES_USER").unwrap_or("".to_string());
+    let postgres_password = env::var("POSTGRES_PASSWORD").unwrap_or("".to_string());
+    let db_host = env::var("DB_HOST").unwrap_or("localhost".to_string());
+    let db_port = env::var("DB_PORT")
+        .unwrap_or("5432".to_string())
+        .to_string();
+    let postgres_db = env::var("POSTGRES_DB").unwrap_or("nocturne".to_string());
+
+    format!("postgresql://{postgres_user}:{postgres_password}@{db_host}:{db_port}/{postgres_db}")
+}