feat(blog::components::check_update) impl consistency_guard, fix handle_create issues

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

@@ -5,10 +5,10 @@ use crate::modules::blog::components::check_update::{
 };
 
 use futures::{
-    future::{join_all, BoxFuture},
     FutureExt,
+    future::{BoxFuture, join_all},
 };
-use sqlx::{query, PgPool, Postgres, Row, Transaction};
+use sqlx::{PgPool, Postgres, Row, Transaction, query};
 use tracing::error;
 
 const DELETE_POST_BY_ID: &str = include_str!("sql/delete_post_by_id.sql");
@@ -16,7 +16,8 @@ const GET_TAG_ID: &str = include_str!("sql/get_tag_id.sql");
 const GET_CATEGORY_ID: &str = include_str!("sql/get_category_id.sql");
 const CREATE_RETURN_TAG_ID: &str = include_str!("sql/create_return_tag_id.sql");
 const CREATE_RETURN_CATEGORY_ID: &str = include_str!("sql/create_return_category_id.sql");
-const CREATE_POST: &str = include_str!("sql/create_post.sql");
+const CREATE_POST_GENERAL: &str = include_str!("sql/create_post_general.sql");
+const CREATE_POST_TAGS: &str = include_str!("sql/create_post_tags.sql");
 
 pub async fn apply_deltas(db_conn: &PgPool, changes: Vec<Change>) {
     let handlers = changes
@@ -47,7 +48,7 @@ async fn handle_move(db: &PgPool, mv: Move) {
         db,
         Delete {
             uuid: mv.uuid,
-            path: mv.from,
+            path: Some(mv.from),
         },
     )
     .await;
@@ -67,7 +68,7 @@ async fn handle_update(db: &PgPool, update: Update) {
         db,
         Delete {
             uuid: update.uuid,
-            path: update.path.clone(),
+            path: Some(update.path.clone()),
         },
     )
     .await;
@@ -143,8 +144,9 @@ async fn handle_create(db: &PgPool, create: Create) {
         .unwrap_or(None);
     let post_status = meta.get("status").map(|v| v.as_str()).unwrap_or(None);
     let post_content = create.payload.content;
+    let post_hash = create.payload.hash;
 
-    if let Err(e) = query(CREATE_POST)
+    if let Err(e) = query(CREATE_POST_GENERAL)
         .bind(post_id)
         .bind(post_identifier)
         .bind(post_title)
@@ -153,7 +155,17 @@ async fn handle_create(db: &PgPool, create: Create) {
         .bind(post_date_updated)
         .bind(post_status)
         .bind(post_content)
+        .bind(post_hash)
         .bind(category_id)
+        .execute(&mut *tx)
+        .await
+    {
+        error!("failed to insert post {}: {}", &identifier, e);
+        return;
+    }
+
+    if let Err(e) = query(CREATE_POST_TAGS)
+        .bind(post_id)
         .bind(tag_ids)
         .execute(&mut *tx)
         .await

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

@@ -17,7 +17,7 @@ pub enum Change {
 #[derive(Debug, Clone)]
 pub struct Delete {
     pub uuid: Uuid,
-    pub path: PathBuf,
+    pub path: Option<PathBuf>,
 }
 
 #[allow(dead_code)]
@@ -56,7 +56,7 @@ impl CreateDelete2UpdateSlot {
         match (&self.create, &self.delete) {
             (Some(create), Some(delete)) => Some(Change::Move(Move {
                 uuid: create.uuid,
-                from: delete.path.clone(),
+                from: delete.path.clone()?,
                 to: create.path.clone(),
                 payload: create.payload.clone(),
             })),

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

@@ -1,28 +1,23 @@
-use std::{
-    env,
-    path::{Path, PathBuf},
-    sync::Arc,
-};
+pub mod apply;
+pub mod changes;
+mod error;
+mod pull;
+pub mod utils;
+
+use std::{env, path::PathBuf, sync::Arc};
 
 use crate::{
     constants::config::general::{default_value, var_name},
-    modules::blog::components::check_update::{
-        apply::apply_deltas, pull::fetch_deltas, utils::expand_path,
+    modules::blog::components::{
+        check_update::{apply::apply_deltas, pull::fetch_deltas},
+        static_rsc::GIT_WORK_DIR,
     },
     scheduler::task_func::AsyncTaskFunc,
 };
 use macros::panic_with_log;
 
 use sqlx::PgPool;
-use tracing::{log::error, trace, warn, Level};
-
-// mod error;
-// mod index;
-mod apply;
-mod changes;
-mod error;
-mod pull;
-mod utils;
+use tracing::{Level, log::error, trace};
 
 #[derive(Debug)]
 struct Args {
@@ -34,22 +29,6 @@ struct Args {
 }
 
 pub fn task(db_connection: PgPool) -> impl AsyncTaskFunc {
-    // init
-    let git_work_dir = {
-        let path = if let Ok(default_log_path) = env::var(var_name::WORK_DIR) {
-            expand_path(default_log_path)
-        } else {
-            let fallback_path = expand_path(default_value::WORK_DIR.to_string());
-            warn!(
-                "unset env var `{}`, using default path: {}/blog_git",
-                var_name::WORK_DIR,
-                fallback_path
-            );
-            fallback_path
-        };
-        Path::new(&path).join("blog_git")
-    };
-
     // clone & check update
     let git_url = match env::var(var_name::BLOG_GIT_URL) {
         Ok(url) => url,
@@ -69,7 +48,7 @@ pub fn task(db_connection: PgPool) -> impl AsyncTaskFunc {
 
     let arc_args = Arc::new(Args {
         db_connection,
-        git_work_dir,
+        git_work_dir: GIT_WORK_DIR.clone(),
         git_url,
         git_remote_name,
         git_remote_branch,

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

@@ -1,17 +1,19 @@
-use std::sync::LazyLock;
-use std::{collections::HashMap, fs::remove_dir_all, path::PathBuf};
+use std::{collections::HashMap, fs::remove_dir_all, path::Path, path::PathBuf};
 
 use crate::{
-    modules::blog::components::check_update::{
-        changes::{Change, Create, CreateDelete2UpdateSlot, Delete, Update},
-        error::Error,
+    modules::blog::components::{
+        check_update::{
+            changes::{Change, Create, CreateDelete2UpdateSlot, Delete, Update},
+            error::Error,
+            utils::extract_post_id,
+        },
+        static_rsc::BLOG_POST_PATH_PATTERN,
     },
     utils::{git, git::FileDelta},
 };
 use markdown::MdFile;
 
 use git2::{Delta, Repository};
-use regex::Regex;
 use tracing::{error, warn};
 use uuid::Uuid;
 
@@ -80,7 +82,7 @@ pub async fn fetch_deltas(
             if let Some(id) = extract_post_id(&MdFile::from_file(file_path)?) {
                 changes.push(Change::Delete(Delete {
                     uuid: id,
-                    path: file_path.clone(),
+                    path: Some(file_path.clone()),
                 }))
             } else {
                 error!("failed to parse uuid for file {}", file_path.display());
@@ -136,18 +138,11 @@ pub async fn fetch_deltas(
     }
 
     // merge adds and deletes deltas to moves
-    let changes = dbg!(merge_changes(changes));
+    let changes = merge_changes(changes);
 
     Ok(changes)
 }
 
-fn extract_post_id(post: &MdFile) -> Option<Uuid> {
-    match post.meta["uuid"].as_str() {
-        Some(id) => Uuid::parse_str(id).ok(),
-        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();
@@ -175,9 +170,6 @@ fn merge_changes(changes: Vec<Change>) -> Vec<Change> {
     merged
 }
 
-static TRACKED_DELTA_RULE: LazyLock<Regex> =
-    LazyLock::new(|| Regex::new(r"posts/\d{4}/[a-zA-Z0-9-]+/index\.md").unwrap());
-
-fn should_delta_tracked(path: &PathBuf) -> bool {
-    TRACKED_DELTA_RULE.is_match(path.to_str().unwrap())
+fn should_delta_tracked(path: &Path) -> bool {
+    BLOG_POST_PATH_PATTERN.is_match(path.to_str().unwrap())
 }

backend/server/src/modules/blog/components/check_update/sql/create_post.sql

@@ -1,7 +0,0 @@
-INSERT INTO posts
-(id, identifier, title, subtitle, date_created, date_updated, status, content, category)
-VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9);
-
-INSERT INTO post_tag (post, tag)
-SELECT $1, tag
-FROM UNNEST($10) AS t(tag);

backend/server/src/modules/blog/components/check_update/sql/create_post_general.sql

@@ -0,0 +1,3 @@
+INSERT INTO posts
+(id, identifier, title, subtitle, date_created, date_updated, status, content, hash, category)
+VALUES ($1, $2, $3, $4, $5, $6, COALESCE($7, 'draft'), $8, $9, $10);

backend/server/src/modules/blog/components/check_update/sql/create_post_tags.sql

@@ -0,0 +1,3 @@
+INSERT INTO post_tag (post, tag)
+SELECT $1, tag
+FROM UNNEST($2) AS t(tag);

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

@@ -7,7 +7,9 @@ use crate::constants::{
 
 use chrono::NaiveDate;
 use dirs::home_dir;
+use markdown::MdFile;
 use tracing::warn;
+use uuid::Uuid;
 
 pub fn expand_path(path: String) -> String {
     if path.starts_with('~') {
@@ -31,6 +33,13 @@ pub fn expand_path(path: String) -> String {
     }
 }
 
+pub fn extract_post_id(post: &MdFile) -> Option<Uuid> {
+    match post.meta["id"].as_str() {
+        Some(id) => Uuid::parse_str(id).ok(),
+        None => None,
+    }
+}
+
 pub fn parse_post_identifier(path: &Path) -> Option<String> {
     if let Some(comp) = path.components().nth_back(1) {
         comp.as_os_str().to_str().map(|s| s.to_string())

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

@@ -0,0 +1,130 @@
+use std::{
+    collections::HashMap,
+    fs,
+    path::{Path, PathBuf},
+    str::FromStr,
+};
+
+use crate::modules::blog::components::{
+    check_update::{
+        changes::{Change, Create, Delete, Update},
+        utils::extract_post_id,
+    },
+    consistency_guard::{error::Error, utils::rec_find_dir_entries},
+    static_rsc::BLOG_POST_PATH_PATTERN,
+};
+use markdown::MdFile;
+
+use sqlx::{FromRow, PgPool, query_as};
+use tracing::error;
+use uuid::Uuid;
+
+const FETCH_POST_HASH: &str = include_str!("sql/fetch_post_hash.sql");
+
+#[derive(Debug, FromRow)]
+struct PostIdHash {
+    id: Uuid,
+    hash: String,
+}
+
+struct PostPath {
+    id: Uuid,
+    path: PathBuf,
+    payload: MdFile,
+}
+
+#[derive(Default)]
+struct DeltaSlot {
+    local: Option<PostPath>,
+    db: Option<PostIdHash>,
+}
+
+impl DeltaSlot {
+    pub fn to_change(&self) -> Option<Change> {
+        match (&self.local, &self.db) {
+            (Some(local), Some(db)) => {
+                if local.payload.hash == db.hash {
+                    None
+                } else {
+                    Some(Change::Update(Update {
+                        uuid: local.id,
+                        path: local.path.clone(),
+                        payload: local.payload.clone(),
+                    }))
+                }
+            }
+            (Some(local), None) => Some(Change::Create(Create {
+                uuid: local.id,
+                path: local.path.clone(),
+                payload: local.payload.clone(),
+            })),
+            (None, Some(db)) => Some(Change::Delete(Delete {
+                uuid: db.id,
+                path: None,
+            })),
+            (None, None) => None,
+        }
+    }
+}
+
+pub async fn compare_db_local_post_hash(
+    db: &PgPool,
+    git_work_dir: &Path,
+) -> Result<Vec<Change>, Error> {
+    let db_post_hashes = query_as::<_, PostIdHash>(FETCH_POST_HASH)
+        .fetch_all(db)
+        .await?;
+    let local_posts = fetch_local_posts(git_work_dir)?;
+    calculate_post_changes(local_posts, db_post_hashes)
+}
+
+fn fetch_local_posts(git_work_dir: &Path) -> Result<Vec<PostPath>, Error> {
+    let mut post_hashes = vec![];
+    let entries = rec_find_dir_entries(&git_work_dir.join("posts"))?;
+
+    for entry in entries {
+        let relative_pth = entry.strip_prefix(git_work_dir).unwrap();
+        if !BLOG_POST_PATH_PATTERN.is_match(relative_pth.to_str().unwrap()) {
+            continue;
+        }
+        let post_str = fs::read_to_string(&entry)?;
+        let post = MdFile::from_str(&post_str)?;
+        let post_id = match extract_post_id(&post) {
+            None => {
+                error!(
+                    "failed to parse uuid for file {}, skipping",
+                    &entry.display()
+                );
+                continue;
+            }
+            Some(v) => v,
+        };
+        post_hashes.push(PostPath {
+            id: post_id,
+            path: entry,
+            payload: post,
+        });
+    }
+
+    Ok(post_hashes)
+}
+
+fn calculate_post_changes(local: Vec<PostPath>, db: Vec<PostIdHash>) -> Result<Vec<Change>, Error> {
+    let mut slots: HashMap<Uuid, DeltaSlot> = HashMap::new();
+    for local_post in local {
+        let id = local_post.id;
+        slots.entry(id).or_default().local = Some(local_post);
+    }
+    for db_post in db {
+        let id = db_post.id;
+        slots.entry(id).or_default().db = Some(db_post);
+    }
+
+    let mut changes = vec![];
+    for slot in slots.values() {
+        if let Some(change) = slot.to_change() {
+            changes.push(change);
+        }
+    }
+    Ok(changes)
+}

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

@@ -0,0 +1,19 @@
+use std::io;
+use std::path::PathBuf;
+
+use thiserror::Error;
+
+#[derive(Error, Debug)]
+pub enum Error {
+    #[error("db error")]
+    Db(#[from] sqlx::Error),
+
+    #[error("io error")]
+    Io(#[from] io::Error),
+
+    #[error("invalid path: {0}")]
+    InvalidPath(PathBuf),
+
+    #[error("invalid md file")]
+    MdFile(#[from] markdown::error::Error),
+}

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

@@ -1 +1,44 @@
+mod checker;
+mod error;
+mod utils;
 
+use std::{path::PathBuf, sync::Arc};
+
+use crate::{
+    modules::blog::components::{
+        check_update::apply::apply_deltas, consistency_guard::checker::compare_db_local_post_hash,
+        static_rsc::GIT_WORK_DIR,
+    },
+    scheduler::task_func::AsyncTaskFunc,
+};
+
+use sqlx::PgPool;
+use tracing::{error, trace};
+
+struct Args {
+    db_connection: PgPool,
+    git_work_dir: PathBuf,
+}
+
+pub fn task(db_connection: PgPool) -> impl AsyncTaskFunc {
+    let args = Arc::new(Args {
+        db_connection,
+        git_work_dir: GIT_WORK_DIR.clone(),
+    });
+    move || Box::pin(workflow(args.clone()))
+}
+
+async fn workflow(args: Arc<Args>) {
+    let db = &args.db_connection;
+    let git_work_dir = &args.git_work_dir;
+
+    let changes = match compare_db_local_post_hash(db, git_work_dir).await {
+        Ok(c) => c,
+        Err(e) => {
+            error!("failed to compare db and local post list: {}", e);
+            return;
+        }
+    };
+    apply_deltas(db, changes).await;
+    trace!("local-database consistency checked");
+}

backend/server/src/modules/blog/components/consistency_guard/sql/fetch_post_hash.sql

@@ -0,0 +1,2 @@
+SELECT id, hash
+FROM posts

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

@@ -0,0 +1,77 @@
+use crate::modules::blog::components::consistency_guard::error::Error;
+use std::{
+    fs,
+    path::{Path, PathBuf},
+};
+
+pub fn rec_find_dir_entries(entry: &Path) -> Result<Vec<PathBuf>, Error> {
+    if !entry.exists() {
+        return Err(Error::InvalidPath(entry.to_path_buf()));
+    }
+
+    let mut result = vec![];
+
+    if !entry.is_dir() {
+        return Ok(vec![entry.to_path_buf()]);
+    }
+
+    for entry in fs::read_dir(entry)? {
+        let entry = entry?;
+        let path = entry.path();
+
+        if path.is_dir() {
+            result.extend(rec_find_dir_entries(&path)?);
+        } else {
+            result.push(path);
+        }
+    }
+
+    Ok(result)
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use std::fs::{self, File};
+    use tempfile::TempDir;
+
+    #[test]
+    fn test_valid_directory() {
+        let temp_dir = TempDir::new().unwrap();
+        let dir_path = temp_dir.path();
+
+        fs::create_dir(dir_path.join("subdir")).unwrap();
+        File::create(dir_path.join("file1.txt")).unwrap();
+        File::create(dir_path.join("subdir/file2.txt")).unwrap();
+
+        let entries = rec_find_dir_entries(dir_path).unwrap();
+        assert_eq!(entries.len(), 2);
+    }
+
+    #[test]
+    fn test_non_existent_path() {
+        let non_existent = PathBuf::from("/non/existent/path");
+        assert!(matches!(
+            rec_find_dir_entries(&non_existent),
+            Err(Error::InvalidPath(_))
+        ));
+    }
+
+    #[test]
+    fn test_single_file() {
+        let temp_dir = TempDir::new().unwrap();
+        let file_path = temp_dir.path().join("test.txt");
+        File::create(&file_path).unwrap();
+
+        let entries = rec_find_dir_entries(&file_path).unwrap();
+        assert_eq!(entries.len(), 1);
+        assert_eq!(entries[0], file_path);
+    }
+
+    #[test]
+    fn test_empty_directory() {
+        let temp_dir = TempDir::new().unwrap();
+        let entries = rec_find_dir_entries(temp_dir.path()).unwrap();
+        assert!(entries.is_empty());
+    }
+}

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

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

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

@@ -0,0 +1,31 @@
+use std::{
+    env,
+    path::{Path, PathBuf},
+    sync::LazyLock,
+};
+
+use crate::{
+    constants::config::general::{default_value, var_name},
+    modules::blog::components::check_update::utils::expand_path,
+};
+
+use regex::Regex;
+use tracing::warn;
+
+pub static BLOG_POST_PATH_PATTERN: LazyLock<Regex> =
+    LazyLock::new(|| Regex::new(r"^posts/\d{4}/[a-zA-Z0-9-]+/index\.md$").unwrap());
+
+pub static GIT_WORK_DIR: LazyLock<PathBuf> = LazyLock::new(|| {
+    let path = if let Ok(default_work_path) = env::var(var_name::WORK_DIR) {
+        expand_path(default_work_path)
+    } else {
+        let fallback_path = expand_path(default_value::WORK_DIR.to_string());
+        warn!(
+            "unset env var `{}`, using default path: {}/blog_git",
+            var_name::WORK_DIR,
+            fallback_path
+        );
+        fallback_path
+    };
+    Path::new(&path).join("blog_git")
+});