feat(scheduler) fix rustc warning

backend/markdown/src/lib.rs

@@ -1,4 +1,4 @@
-mod error;
+pub mod error;
 mod utils;
 
 use std::{fs, path::Path, str::FromStr};
@@ -8,12 +8,12 @@ use crate::{
     utils::front_matter::{parse_front_matter, split_md_front_matter},
 };
 
-use base64::{Engine, engine::general_purpose};
+use base64::{engine::general_purpose, Engine};
 use serde::Serialize;
 use serde_json::Value;
 use sha2::{Digest, Sha256};
 
-#[derive(Default, Debug, Serialize)]
+#[derive(Default, Debug, Serialize, Clone)]
 pub struct MdFile {
     pub meta: Value,
     pub content: String,

backend/server/src/constants/config.rs

@@ -1,3 +1,5 @@
+// todo: rebuild with static OnceCell
+
 // Database related
 pub mod database {
     pub mod var_name {

backend/server/src/modules/blog/api/get_post/get_post_by_i64_id.sql

@@ -0,0 +1 @@
+SELECT 

backend/server/src/modules/blog/api/get_post/mod.rs

@@ -1,9 +1,9 @@
 use macros::dev_consume;
 
 use axum::{
+    Json,
     extract::{Path, State},
     response::IntoResponse,
-    Json,
 };
 use serde_json::json;
 use sqlx::PgPool;

backend/server/src/modules/blog/components/check_update/changes.rs

@@ -0,0 +1,67 @@
+use std::path::PathBuf;
+
+use markdown::MdFile;
+
+use uuid::Uuid;
+
+#[allow(dead_code)]
+#[derive(Debug)]
+pub enum Change {
+    Delete(Delete),
+    Move(Move),
+    Update(Update),
+    Create(Create),
+}
+
+#[allow(dead_code)]
+#[derive(Debug, Clone)]
+pub struct Delete {
+    pub uuid: Uuid,
+    pub path: PathBuf,
+}
+
+#[allow(dead_code)]
+#[derive(Debug, Clone)]
+pub struct Move {
+    pub uuid: Uuid,
+    pub from: PathBuf,
+    pub to: PathBuf,
+    pub payload: MdFile,
+}
+
+#[allow(dead_code)]
+#[derive(Debug, Clone)]
+pub struct Update {
+    pub uuid: Uuid,
+    pub payload: MdFile,
+}
+
+#[allow(dead_code)]
+#[derive(Debug, Clone)]
+pub struct Create {
+    pub uuid: Uuid,
+    pub path: PathBuf,
+    pub payload: MdFile,
+}
+
+#[derive(Debug, Clone, Default)]
+pub struct CreateDelete2UpdateSlot {
+    pub create: Option<Create>,
+    pub delete: Option<Delete>,
+}
+
+impl CreateDelete2UpdateSlot {
+    pub fn to_change(&self) -> Option<Change> {
+        match (&self.create, &self.delete) {
+            (Some(create), Some(delete)) => Some(Change::Move(Move {
+                uuid: create.uuid,
+                from: delete.path.clone(),
+                to: create.path.clone(),
+                payload: create.payload.clone(),
+            })),
+            (Some(create), None) => Some(Change::Create(create.clone())),
+            (None, Some(delete)) => Some(Change::Delete(delete.clone())),
+            (None, None) => None,
+        }
+    }
+}

backend/server/src/modules/blog/components/check_update/error.rs

@@ -1,7 +1,13 @@
+use markdown::error::Error as MarkdownError;
+
 use thiserror::Error;
 
+#[allow(dead_code)]
 #[derive(Error, Debug)]
-pub enum PostIdxError {
+pub enum Error {
+    #[error("markdown file error")]
+    MdFile(#[from] MarkdownError),
+
     #[error("missing essential field `{}` for file `{}`", field, filename)]
     MissingField { field: String, filename: String },
 
@@ -46,4 +52,7 @@ pub enum PostIdxError {
         id: String,
         msg: String,
     },
+
+    #[error("git error")]
+    Git2(#[from] git2::Error),
 }

backend/server/src/modules/blog/components/check_update/mod.rs

@@ -1,26 +1,29 @@
-use crate::constants::config::general::{default_value, var_name};
-use macros::panic_with_log;
-use std::sync::Arc;
 use std::{
     env,
     path::{Path, PathBuf},
+    sync::Arc,
 };
 
-use crate::modules::blog::components::check_update::{
+use crate::{
-    // index::drop_index,
+    constants::config::general::{default_value, var_name},
-    // utils::{expand_path, index_files, search_md},
+    modules::blog::components::check_update::{pull::fetch_deltas, utils::expand_path},
-    utils::expand_path,
+    scheduler::task_func::AsyncTaskFunc,
 };
-use crate::scheduler::task_func::AsyncTaskFunc;
+use macros::panic_with_log;
+
 use sqlx::PgPool;
-use tracing::{info, warn, Level};
+use tracing::{trace, warn, Level};
 
 // mod error;
 // mod index;
+mod changes;
+mod error;
+mod pull;
 mod utils;
 
 #[derive(Debug)]
 struct Args {
+    db_connection: PgPool,
     git_work_dir: PathBuf,
     git_url: String,
     git_remote_name: String,
@@ -45,7 +48,6 @@ pub fn task(db_connection: PgPool) -> impl AsyncTaskFunc {
     };
 
     // clone & check update
-
     let git_url = match env::var(var_name::BLOG_GIT_URL) {
         Ok(url) => url,
         Err(e) => {
@@ -63,16 +65,33 @@ pub fn task(db_connection: PgPool) -> impl AsyncTaskFunc {
         .unwrap_or(default_value::BLOG_GIT_REMOTE_BRANCH.to_string());
 
     let arc_args = Arc::new(Args {
+        db_connection,
         git_work_dir,
         git_url,
         git_remote_name,
         git_remote_branch,
     });
 
-    move || Box::pin(test(arc_args.clone()))
+    move || Box::pin(workflow(arc_args.clone()))
 }
 
-async fn test(args: Arc<Args>) {
+async fn workflow(args: Arc<Args>) {
-    info!("test");
+    let deltas = match fetch_deltas(
-    dbg!(&args);
+        &args.git_url,
+        &args.git_remote_name,
+        &args.git_remote_branch,
+        &args.git_work_dir,
+    )
+    .await
+    {
+        Ok(updates) => updates,
+        Err(_) => return,
+    };
+
+    if deltas.is_empty() {
+        trace!("blog git already up-to-date");
+        return;
+    }
+
+    // todo: load delta to database
 }

backend/server/src/modules/blog/components/check_update/pull.rs

@@ -0,0 +1,153 @@
+use std::collections::HashMap;
+use std::path::PathBuf;
+
+use crate::{
+    modules::blog::components::check_update::{
+        changes::{Change, Create, Delete, Update},
+        error::Error,
+    },
+    utils::{git, git::FileDelta},
+};
+use markdown::MdFile;
+
+use crate::modules::blog::components::check_update::changes::CreateDelete2UpdateSlot;
+use git2::{Delta, Repository};
+use tracing::error;
+use uuid::Uuid;
+
+pub async fn fetch_deltas(
+    git_url: &str,
+    git_remote_name: &str,
+    git_remote_branch: &str,
+    git_work_dir: &PathBuf,
+) -> Result<Vec<Change>, Error> {
+    let repo = match Repository::open(git_work_dir) {
+        Ok(repo) => repo,
+        Err(_) => match Repository::clone(git_url, git_work_dir) {
+            Ok(repo) => repo,
+            Err(e) => {
+                error!("failed to clone blog content git({}): {}", &git_url, e);
+                return Err(e.into());
+            }
+        },
+    };
+
+    let mut remote = repo.find_remote(git_remote_name)?;
+    let fetched_commit = git::do_fetch(&repo, &[git_remote_branch], &mut remote)?;
+    let commit_deltas = git::calculate_diff(
+        &repo,
+        &repo.reference_to_annotated_commit(&repo.head()?)?,
+        &fetched_commit,
+    )?;
+
+    // dbg!(&commit_deltas);
+    let mut changes: Vec<Change> = vec![];
+    let mut non_delete_deltas: Vec<FileDelta> = vec![];
+
+    // process deleted deltas before merging commit to prevent "file not found" os error
+    for delta in commit_deltas.into_iter() {
+        if delta.status == Delta::Deleted {
+            let file_path = &git_work_dir.join(
+                delta
+                    .new_path
+                    .expect("cannot parse modified post `new_path`"),
+            );
+            if let Some(id) = extract_post_id(&MdFile::from_file(file_path)?) {
+                changes.push(Change::Delete(Delete {
+                    uuid: id,
+                    path: file_path.clone(),
+                }))
+            } else {
+                error!("failed to parse uuid for file {}", file_path.display());
+                // todo: add to unresolved error list
+            }
+        } else {
+            non_delete_deltas.push(delta);
+        }
+    }
+
+    git::do_merge(&repo, git_remote_branch, fetched_commit)?;
+
+    // process adds and modifies deltas
+    for delta in non_delete_deltas.into_iter() {
+        let file_path = &git_work_dir.join(
+            delta
+                .new_path
+                .expect("cannot parse modified post `new_path`"),
+        );
+        match delta.status {
+            Delta::Added => {
+                let md_file = MdFile::from_file(file_path)?;
+                if let Some(id) = extract_post_id(&md_file) {
+                    changes.push(Change::Create(Create {
+                        uuid: id,
+                        path: file_path.clone(),
+                        payload: md_file,
+                    }))
+                } else {
+                    error!("failed to parse uuid for file {}", file_path.display());
+                    // todo: add to unresolved error list
+                }
+            }
+            Delta::Modified => {
+                let md_file = MdFile::from_file(file_path)?;
+                if let Some(id) = extract_post_id(&md_file) {
+                    changes.push(Change::Update(Update {
+                        uuid: id,
+                        payload: md_file,
+                    }))
+                } else {
+                    error!("failed to parse uuid for file {}", file_path.display());
+                    // todo: add to unresolved error list
+                }
+            }
+            Delta::Deleted => continue,
+            _ => {
+                error!("unsupported delta type on file {}", file_path.display());
+                continue;
+            }
+        }
+    }
+
+    // merge adds and deletes deltas to moves
+    let changes = dbg!(merge_changes(changes));
+
+    Ok(changes)
+}
+
+fn extract_post_id(post: &MdFile) -> Option<Uuid> {
+    match post.meta["uuid"].as_str() {
+        Some(id) => match Uuid::parse_str(id) {
+            Ok(id) => Some(id),
+            Err(_) => None,
+        },
+        None => None,
+    }
+}
+
+fn merge_changes(changes: Vec<Change>) -> Vec<Change> {
+    let mut slots: HashMap<Uuid, CreateDelete2UpdateSlot> = HashMap::new();
+    let mut merged: Vec<Change> = Vec::new();
+
+    for change in changes {
+        match change {
+            Change::Create(create) => {
+                let id = create.uuid;
+                slots.entry(id).or_default().create = Some(create);
+            }
+            Change::Delete(delete) => {
+                let id = delete.uuid;
+                slots.entry(id).or_default().delete = Some(delete);
+            }
+            _ => merged.push(change),
+        }
+    }
+
+    for slot in slots.values() {
+        if let Some(change) = slot.to_change() {
+            merged.push(change);
+        }
+    }
+
+    merged
+}

backend/server/src/modules/blog/components/check_update/utils.rs

@@ -8,48 +8,6 @@ use chrono::{DateTime, FixedOffset, ParseResult};
 use dirs::home_dir;
 use tracing::warn;
 
-// pub fn search_md(entry: PathBuf) -> Vec<PathBuf> {
-//     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)
-//                 })
-//                 .collect()
-//         } else {
-//             vec![]
-//         }
-//     } else if entry.extension().is_some_and(|ext| ext == "md") {
-//         vec![entry]
-//     } else {
-//         vec![]
-//     }
-// }
-
-// pub fn index_files(files: &[PathBuf], db_conn: &Pool<Postgres>) {
-//     let rt = Runtime::new().unwrap();
-//     rt.block_on(async {
-//         let files_clone = files.to_owned();
-//         let handlers = files_clone
-//             .into_iter()
-//             .map(|file: PathBuf| {
-//                 let conn_clone = db_conn.clone();
-//                 spawn(async move {
-//                     if let Err(e) = index_file(&file, &conn_clone).await {
-//                         error!("failed to index markdown file `{}`: {}", file.display(), e);
-//                     }
-//                 })
-//             })
-//             .collect::<Vec<_>>();
-//
-//         block_on(join_all(handlers));
-//     });
-// }
-
 pub fn expand_path(path: String) -> String {
     if path.starts_with('~') {
         home_dir()
@@ -75,7 +33,3 @@ pub fn expand_path(path: String) -> String {
 pub fn parse_date(date_str: &str) -> ParseResult<DateTime<FixedOffset>> {
     DateTime::parse_from_str(format!("{}T120000", date_str).as_str(), DEFAULT_DATE_FORMAT)
 }
-
-// pub fn get_md_file_basename(f_name: &str) -> String {
-//     f_name.replace(".md", "")
-// }

backend/server/src/modules/blog/components/consistency_guard/mod.rs

@@ -0,0 +1 @@
+

backend/server/src/modules/blog/components/mod.rs

@@ -1,2 +1,3 @@
 pub mod check_update;
+mod consistency_guard;
 mod rag;

backend/server/src/modules/blog/components/rag/mod.rs

@@ -0,0 +1 @@
+

backend/server/src/modules/blog/mod.rs

@@ -1,6 +1,8 @@
 mod api;
 mod components;
 
+use crate::modules::blog::components::check_update;
+use crate::scheduler::tasks::async_basic::AsyncBasic;
 use crate::{
     modules::{
         blog::{
@@ -16,9 +18,6 @@ use crate::{
     scheduler::tasks::CronTask,
 };
 
-use crate::modules::blog::components::check_update;
-use crate::scheduler::tasks::async_basic::AsyncBasic;
-use crate::scheduler::tasks::basic::BasicTask;
 use axum::{routing::get, Router};
 use sqlx::PgPool;
 
@@ -46,19 +45,11 @@ impl Module for Blog {
 
     fn get_cron_tasks(&self) -> Vec<(&str, Box<dyn CronTask>)> {
         let db_handler = self.db_handler.clone();
-        vec![
+        vec![(
-            (
             "Blog Check Update",
-                AsyncBasic::new(check_update::task(db_handler.clone()), "* * * * * *")
+            AsyncBasic::new(check_update::task(db_handler.clone()), "*/5 * * * * *")
                 .unwrap()
                 .to_task(),
-            ),
+        )]
-            (
-                "Sync test task",
-                BasicTask::new(|| println!("test lol"), "* * * * * *")
-                    .unwrap()
-                    .to_task(),
-            ),
-        ]
     }
 }

backend/server/src/modules/blog/schema.sql

@@ -2,20 +2,20 @@ CREATE TABLE tags
 (
     id          SERIAL PRIMARY KEY,
     name        VARCHAR(20) UNIQUE NOT NULL,
-    description TEXT
+    description TEXT DEFAULT ''
 );
 
 CREATE TABLE categories
 (
     id          SERIAL PRIMARY KEY,
     name        VARCHAR(20) UNIQUE NOT NULL,
-    description TEXT
+    description TEXT DEFAULT ''
 );
 
 CREATE TABLE posts
 (
-    id             SERIAL PRIMARY KEY,
+    id             UUID UNIQUE                       NOT NULL,
-    identifier     VARCHAR(50) UNIQUE                NOT NULL,
+    identifier     VARCHAR(100) UNIQUE               NOT NULL,
     title          VARCHAR(255)                      NOT NULL,
     subtitle       VARCHAR(255)                      NOT NULL,
     published_date DATE                              NOT NULL,
@@ -27,13 +27,13 @@ CREATE TABLE posts
 
 CREATE TABLE post_tag
 (
-    post SERIAL REFERENCES posts (id),
+    post UUID REFERENCES posts (id),
     tag  SERIAL REFERENCES tags (id)
 );
 
 CREATE TABLE metadata
 (
-    pid        SERIAL REFERENCES posts (id) NOT NULL,
+    pid        UUID REFERENCES posts (id) NOT NULL,
     meta_key   VARCHAR(50)                NOT NULL,
     meta_value TEXT                       NOT NULL,
 

backend/server/src/modules/mod.rs

@@ -2,7 +2,7 @@ use std::{env, num::ParseIntError};
 
 use crate::{
     constants::config::server::{default_value, var_name},
-    scheduler::{tasks::CronTask, Scheduler},
+    scheduler::{Scheduler, tasks::CronTask},
 };
 use macros::error_panic;
 

backend/server/src/scheduler/mod.rs

@@ -20,8 +20,6 @@ pub struct Scheduler {
 impl Scheduler {
     /// Start all tasks in the scheduler
     pub async fn start(&self) {
-        dbg!("entering");
-
         let fut = self
             .task_pool
             .clone()

backend/server/src/scheduler/tasks/mod.rs

@@ -2,7 +2,10 @@ use std::pin::Pin;
 
 use chrono::{DateTime, Utc};
 
+#[allow(dead_code)]
 pub mod async_basic;
+
+#[allow(dead_code)]
 pub mod basic;
 
 pub trait CronTask: Send + Sync {

backend/server/src/scheduler/utils.rs

@@ -3,7 +3,7 @@ use std::{sync::Arc, time::Duration};
 use crate::scheduler::{error::SchedulerError, tasks::CronTask};
 
 use chrono::{DateTime, Utc};
-use tokio::time::{sleep, Sleep};
+use tokio::time::{Sleep, sleep};
 
 pub fn duration_from_now(future: DateTime<Utc>) -> Duration {
     future

backend/server/src/utils/git.rs

@@ -1,4 +1,4 @@
-/// git merge
+/// git operations
 /// based on https://github.com/rust-lang/git2-rs/blob/master/examples/pull.rs
 /// https://github.com/rust-lang/git2-rs/commit/f3b87baed1e33d6c2d94fe1fa6aa6503a071d837
 ///
@@ -16,7 +16,7 @@ pub struct FileDelta {
     pub status: Delta,
 }
 
-fn do_fetch<'a>(
+pub fn do_fetch<'a>(
     repo: &'a Repository,
     refs: &[&str],
     remote: &'a mut Remote,
@@ -40,7 +40,6 @@ fn fast_forward(
         None => String::from_utf8_lossy(lb.name_bytes()).to_string(),
     };
     let msg = format!("fast-forward({}) -> {}", name, rc.id());
-    // println!("{}", msg);
     lb.set_target(rc.id(), &msg)?;
     repo.set_head(&name)?;
     repo.checkout_head(Some(build::CheckoutBuilder::default().force()))?;
@@ -83,11 +82,11 @@ fn normal_merge(
     Ok(())
 }
 
-fn do_merge<'a>(
+pub fn do_merge<'a>(
     repo: &'a Repository,
     remote_branch: &str,
     fetch_commit: AnnotatedCommit<'a>,
-) -> Result<Vec<FileDelta>, git2::Error> {
+) -> Result<(), git2::Error> {
     let analysis = repo.merge_analysis(&[&fetch_commit])?;
     let head_commit = repo.reference_to_annotated_commit(&repo.head()?)?;
 
@@ -116,10 +115,11 @@ fn do_merge<'a>(
     } else if analysis.0.is_normal() {
         normal_merge(repo, &head_commit, &fetch_commit)?;
     }
-    calculate_diff(repo, &head_commit, &fetch_commit)
+
+    Ok(())
 }
 
-fn calculate_diff(
+pub fn calculate_diff(
     repo: &Repository,
     local: &AnnotatedCommit,
     remote: &AnnotatedCommit,
@@ -149,6 +149,7 @@ fn calculate_diff(
     Ok(updated_files.into_iter().collect())
 }
 
+#[allow(dead_code)]
 pub fn sync(
     remote_name: &str,
     remote_branch: &str,
@@ -156,5 +157,11 @@ pub fn sync(
 ) -> Result<Vec<FileDelta>, git2::Error> {
     let mut remote = repo.find_remote(remote_name)?;
     let fetch_commit = do_fetch(repo, &[remote_branch], &mut remote)?;
-    do_merge(repo, remote_branch, fetch_commit)
+    let delta = calculate_diff(
+        repo,
+        &repo.reference_to_annotated_commit(&repo.head()?)?,
+        &fetch_commit,
+    );
+    do_merge(repo, remote_branch, fetch_commit)?;
+    delta
 }