ref(backend) start of refactor

This commit is contained in:
lumither committed 2024-09-04 23:18:41 -07:00
1 parent 3c1fc34c2a
commit c1ba41ea16
36 files changed
+183 -42

No files matched your search

@@ -0,0 +1 @@
pub mod refresh;
@@ -0,0 +1,61 @@
use thiserror::Error;
#[derive(Error, Debug)]
pub enum PostIdxError {
#[error("Failed to parse UUID <{}>: {}", id, err_msg)]
InvalidUUID { id: String, err_msg: String },
#[error(
"Failed to parse `last update time` for <{}> when creating post index: {}",
id,
err_msg
)]
InvalidLastUpdate { id: String, err_msg: String },
#[error(
"Failed to parse `first update time` for <{}> when creating post index: {}",
id,
err_msg
)]
InvalidFirstUpdate { id: String, err_msg: String },
#[error(
"Failed to parse `tag` for <{}> when creating post index: <{}>: {}",
id,
invalid_string,
err_msg
)]
InvalidTagString {
id: String,
invalid_string: String,
err_msg: String,
},
#[error(
"Database error on writing `{}` to table `{}`: <{}>: {}",
data_desc,
db_table,
id,
err_msg
)]
DBWriteFailure {
data_desc: String,
db_table: String,
id: String,
err_msg: String,
},
#[error(
"Database error on Reading `{}` from table `{}`: <{}>: {}",
data_desc,
db_table,
id,
err_msg
)]
DBReadFailure {
data_desc: String,
db_table: String,
id: String,
err_msg: String,
},
}
+3
View File
@@ -0,0 +1,3 @@
mod errors;
pub mod refresh;
mod utils;
+192
View File
@@ -0,0 +1,192 @@
use std::path::Path;
use std::str::FromStr;
use std::{env, fs};
use axum::extract::{Query, State};
use axum::http::StatusCode;
use axum::Json;
use futures::{stream, StreamExt};
use serde::Deserialize;
use serde_json::{json, Map, Value};
use sha2::{Digest, Sha256};
use sqlx::{query, PgPool};
use tracing::error;
use uuid::Uuid;
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)
}
}
/// 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> {
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(())
}
@@ -0,0 +1,288 @@
use base64::{Engine, engine::general_purpose};
use chrono::DateTime;
use serde_json::{Map, Value};
use sqlx::{PgPool, query, Row};
use uuid::Uuid;
use crate::api::post::errors::refresh::PostIdxError;
use crate::constants::GLOBAL_TIME_FORMAT;
pub async fn base(
db_connection: &PgPool,
post: &Map<String, Value>,
post_id: Uuid,
meta: &Value,
) -> Result<(), PostIdxError> {
let title = post["title"].as_str();
let summary = post
.get("summary")
.and_then(|summary| summary.as_str())
.unwrap_or("");
let content = post["content"].as_str();
let last_update =
match DateTime::parse_from_str(post["last_update"].as_str().unwrap(), GLOBAL_TIME_FORMAT) {
Ok(time) => time,
Err(e) => {
return Err(PostIdxError::InvalidLastUpdate {
id: post_id.to_string(),
err_msg: e.to_string(),
});
}
};
let first_update = match DateTime::parse_from_str(
post["first_update"].as_str().unwrap(),
GLOBAL_TIME_FORMAT,
) {
Ok(time) => time,
Err(e) => {
return Err(PostIdxError::InvalidFirstUpdate {
id: post_id.to_string(),
err_msg: e.to_string(),
})
}
};
let sub_title = meta
.get("sub_title")
.and_then(|sub_title| sub_title.as_str())
.unwrap_or("");
let category = meta
.get("category")
.and_then(|category| category.as_str())
.unwrap_or("N/A");
let header_img = meta
.get("header_img")
.and_then(|header_img| header_img.as_str())
.unwrap_or("");
match query(
r##"
INSERT INTO Post (post_id, title, summary, content, last_update, first_update, sub_title, category, header_img)
VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9)
ON CONFLICT (post_id) DO
UPDATE SET
title = EXCLUDED.title,
summary = EXCLUDED.summary,
content = EXCLUDED.content,
last_update = EXCLUDED.last_update,
first_update = EXCLUDED.first_update,
sub_title = EXCLUDED.sub_title,
category = EXCLUDED.category,
header_img = EXCLUDED.header_img;
"##,
)
.bind(post_id)
.bind(title)
.bind(summary)
.bind(content)
.bind(last_update)
.bind(first_update)
.bind(sub_title)
.bind(category)
.bind(header_img)
.execute(db_connection)
.await
{
Ok(_) => {}
Err(e) => {
return Err(
PostIdxError::DBWriteFailure {
data_desc: "post base information".to_string(),
db_table: "Post".to_string(),
id: post_id.to_string(),
err_msg: e.to_string()
}
)
}
}
Ok(())
}
pub async fn tag(db_connection: &PgPool, post_id: Uuid, meta: &Value) -> Result<(), PostIdxError> {
if let Some(tags_string) = meta.get("tags").to_owned() {
// parse tags string
let parsed: Value = match serde_json::from_str(tags_string.as_str().unwrap()) {
Ok(value) => value,
Err(e) => {
return Err(PostIdxError::InvalidTagString {
id: post_id.to_string(),
invalid_string: tags_string.to_string().to_string(),
err_msg: e.to_string(),
})
}
};
let string_array = parsed.as_array().unwrap();
let tags: Vec<String> = string_array
.iter()
.map(|value| value.as_str().unwrap().to_string())
.collect();
// remove old record
if let Err(e) = query("DELETE FROM Tag WHERE post_id = $1")
.bind(post_id)
.execute(db_connection)
.await
{
return Err(PostIdxError::DBWriteFailure {
data_desc: "post tag information".to_string(),
db_table: "Tag".to_string(),
id: post_id.to_string(),
err_msg: e.to_string(),
});
}
// add new record
for tag in tags {
match query("INSERT INTO Tag VALUES ($1, $2);")
.bind(post_id)
.bind(tag)
.execute(db_connection)
.await
{
Ok(_) => {}
Err(e) => {
return Err(PostIdxError::DBWriteFailure {
data_desc: "post tag information".to_string(),
db_table: "Tag".to_string(),
id: post_id.to_string(),
err_msg: e.to_string(),
})
}
}
}
}
Ok(())
}
/// # Arguments
///
/// * `db_connection`: database connection
/// * `post_id`: UUID of the post
/// * `hash`: Hex String
///
/// # Returns
/// Result<bool, PostIdxError>
/// - `bool`: if the hash is different from database, return true, else false
pub async fn hash(
db_connection: &PgPool,
post_id: Uuid,
hash: Vec<u8>,
) -> Result<bool, PostIdxError> {
let base64_form = general_purpose::STANDARD.encode(hash);
match query("SELECT hash FROM Hash WHERE post_id = $1")
.bind(post_id)
.fetch_optional(db_connection)
.await
{
Ok(row) => {
let db_record: String = match row {
None => "".to_string(),
Some(row) => row.try_get("hash").unwrap_or("".to_string()),
};
if base64_form == db_record {
Ok(false)
} else {
if let Err(e) = query(
r#"
INSERT INTO Hash (post_id, hash) VALUES ($1, $2)
ON CONFLICT (post_id) DO UPDATE SET hash = EXCLUDED.hash"#,
)
.bind(post_id)
.bind(&base64_form)
.execute(db_connection)
.await
{
return Err(PostIdxError::DBWriteFailure {
data_desc: "post hash".to_string(),
db_table: "Hash".to_string(),
id: post_id.to_string(),
err_msg: e.to_string(),
});
}
Ok(true)
}
}
Err(e) => Err(PostIdxError::DBReadFailure {
data_desc: "post hash".to_string(),
db_table: "Hash".to_string(),
id: post_id.to_string(),
err_msg: e.to_string(),
}),
}
}
pub async fn meta(db_connection: &PgPool, post_id: Uuid, meta: &Value) -> Result<(), PostIdxError> {
let mut err_trails = 5;
loop {
if let Err(e) = query(
format!(
r#"
DO
$$
DECLARE
_post_id UUID := '{}';
_meta_cols JSONB := '{}';
_exists BOOLEAN;
_col_name TEXT;
_col_value TEXT;
_sql TEXT;
_insert_columns TEXT := 'post_id';
_insert_values TEXT := quote_literal(_post_id);
_update_set TEXT := '';
_blacklist TEXT[] := ARRAY['tags'];
BEGIN
SELECT EXISTS(SELECT 1 FROM meta WHERE post_id = _post_id) INTO _exists;
FOR _col_name, _col_value IN SELECT key, value FROM jsonb_each_text(_meta_cols)
LOOP
IF _col_name = ANY(_blacklist) THEN
CONTINUE;
END IF;
IF NOT EXISTS (SELECT 1
FROM information_schema.columns
WHERE table_name = 'meta'
AND column_name = _col_name
-- may change in the future
AND table_schema = 'public') THEN
_sql := 'ALTER TABLE meta ADD COLUMN ' || quote_ident(_col_name) || ' TEXT';
EXECUTE _sql;
END IF;
_insert_columns := _insert_columns || ', ' || quote_ident(_col_name);
_insert_values := _insert_values || ', ' || quote_literal(_col_value);
_update_set := _update_set || ', ' || quote_ident(_col_name) || ' = ' || quote_literal(_col_value);
END LOOP;
_update_set := substring(_update_set FROM 3);
IF _exists THEN
_sql := 'UPDATE meta SET ' || _update_set || ' WHERE post_id = ' || quote_literal(_post_id);
EXECUTE _sql;
ELSE
_sql := 'INSERT INTO meta (' || _insert_columns || ') VALUES (' || _insert_values || ')';
EXECUTE _sql;
END IF;
END;
$$ LANGUAGE plpgsql;
"#, post_id.to_string().as_str(), meta.to_string()
).as_str()
)
.execute(db_connection)
.await {
if err_trails >= 0 {
err_trails -= 1;
continue
} else {
return Err(PostIdxError::DBWriteFailure {
data_desc: "post meta data".to_string(),
db_table: "Meta".to_string(),
id: post_id.to_string(),
err_msg: e.to_string(),
})
}
} else {
return Ok(())
}
}
}
+2
View File
@@ -0,0 +1,2 @@
pub mod build_post_idx;
pub mod post;
+120
View File
@@ -0,0 +1,120 @@
use std::{env, fs};
use std::error::Error;
use std::process::Command;
use std::str;
use std::str::FromStr;
use chrono::{DateTime, Utc};
use serde_json::{Map, Value};
use tracing::{error, warn};
use uuid::Uuid;
use crate::constants::GLOBAL_TIME_FORMAT;
use crate::markdown::meta::parse_meta;
pub fn from_path(path: &str) -> Result<Map<String, Value>, Box<dyn Error>> {
let mut res = Map::new();
// field `content`
let md_content = fs::read_to_string(path)?;
res.insert("content".to_string(), Value::from(md_content.to_owned()));
// field `meta`
let meta = parse_meta(&md_content);
let meta_map: Map<String, Value> = meta
.clone()
.into_iter()
.map(|(k, v)| (k, Value::from(v)))
.collect();
res.insert("meta".to_string(), Value::from(meta_map));
// field `post_id`
let post_id = match meta.get("id") {
None => Uuid::new_v4(),
Some(id) => match Uuid::from_str(id.as_str()) {
Ok(id) => id,
Err(e) => {
error!("Failed to parse UUID <{}>: {}", id.as_str(), e.to_string());
return Err(Box::new(e));
}
},
};
res.insert("post_id".to_string(), Value::from(post_id.to_string()));
// field `title`
let title = match meta.get("title") {
None => "Title Not Found".to_string(),
Some(title) => title.to_owned(),
};
res.insert("title".to_string(), Value::from(title));
// field `summary`
let summary = match meta.get("summary") {
None => "".to_string(),
Some(summary) => summary.to_owned(),
};
res.insert("summary".to_string(), Value::from(summary));
// field `last_update`
let git_output = match Command::new("git")
.arg("--no-pager")
.arg("log")
.arg("-1")
.arg("--format=%cI")
.arg("--")
.current_dir(env::var("GIT_WORK_DIR")?)
.arg(path)
.output()
{
Ok(res) => res,
Err(e) => {
error!("Failed to execute `git` command: {}", e.to_string());
return Err(Box::new(e));
}
};
let last_update: DateTime<Utc> = if git_output.status.success() {
chrono::DateTime::parse_from_str(
str::from_utf8(&git_output.stdout).unwrap().trim_end(),
GLOBAL_TIME_FORMAT,
)
.unwrap_or_else(|_| DateTime::from(Utc::now()))
.to_utc()
} else {
Utc::now().to_utc()
};
res.insert(
"last_update".to_string(),
Value::from(last_update.format(GLOBAL_TIME_FORMAT).to_string()),
);
// field `first_update`
let first_update: DateTime<Utc> = match meta.get("date") {
None => {
warn!(
"Failed to parse first update time for post {}, using current time instead",
post_id.to_string()
);
Utc::now().to_utc()
}
Some(date) => DateTime::from(
match DateTime::parse_from_str(format!("{}T000000", date).as_str(), "%Y-%m-%d%zT%H%M%S")
{
Ok(time) => time,
Err(e) => {
warn!(
"Failed to parse first update time for post {}, using current time instead: {}",
post_id.to_string(),
e.to_string()
);
DateTime::from(Utc::now())
}
},
),
};
res.insert(
"first_update".to_string(),
Value::from(first_update.format(GLOBAL_TIME_FORMAT).to_string()),
);
Ok(res)
}