29 Commits
Author SHA1 Message Date
lumither 6a6af722c3 feat(blog::components::check_update, README.md) code cleanup, readme update 2025-04-13 03:59:40 -07:00
lumither 512197d87a feat(blog::components::check_update) remove unnecessary import 2025-04-13 03:16:50 -07:00
lumither 3c16bf5100 feat(blog::components::check_update) impl handle_move & handle_update 2025-04-13 03:15:48 -07:00
lumither ff4245a856 feat(blog::components::check_update) impl handle_delete 2025-04-13 03:08:17 -07:00
lumither 9270abc4b7 fix(blog::components::check_update) remove the use of deprecated api 2025-04-12 23:31:52 -07:00
lumither b11b19b0f5 feat(blog::components::check_update) impl handle_create() 2025-04-12 23:30:22 -07:00
lumither 7d988d380b feat(blog::components::check_update) remove redundant error types 2025-04-10 21:53:18 -07:00
lumither 8ba348a93e fix(blog::components::check_update) handle local-remote git url mismatch 2025-04-10 21:51:12 -07:00
lumither 10f14312bd feat(blog::components::check_update) add apply.rs 2025-04-09 19:20:50 -07:00
lumither 27d2c86106 feat(blog/schema.sql) update 2025-04-09 17:37:54 -07:00
lumither 12d9ddf8b6 feat(scheduler) fix rustc warning 2025-04-05 18:37:47 -07:00
lumither ac8a2359b0 feat(blog::components::check+update) update 2025-04-05 18:37:23 -07:00
lumither 40abec2e90 feat(blog::components::check+update) update 2025-04-02 13:48:20 -07:00
lumither 1038351f4a feat(markdown) add from_str 2025-03-26 19:52:19 -07:00
lumither b748764d79 feat(markdown) remove id check & extraction 2025-03-26 19:44:47 -07:00
lumither ac5f30151f feat(scheduler) remove terminate() etc. impl 2025-03-18 12:02:44 -07:00
lumither d54996b690 feat(client) add client 2025-03-13 16:16:50 -07:00
lumither 7b84a63a14 feat(backend::scheduler) deadlock spotted 2025-03-12 18:34:49 -07:00
lumither 7cc44522fa feat(backend::scheduler) fix runtime issue 2025-03-07 23:52:44 -08:00
lumither 2045d380c1 feat(makefile) cleanup 2025-03-06 16:08:00 -08:00
lumither dc781bc2ef feat(backend) module rename & import improve 2025-03-06 00:48:35 -08:00
lumither 2e32297d5e feat(backend) fix deadlock issue, code cleanup 2025-03-05 23:38:40 -08:00
lumither 5e8b1ff35d feat(backend) scheduler rewrite (deadlock for now) 2025-03-05 13:52:07 -08:00
lumither 0c6325d5f2 feat(backend) cargo fmt & migration to rust 2024 2025-02-23 17:39:36 -08:00
lumither ca86dd236e feat(backend) update (unfinished) 2025-02-22 01:15:46 -08:00
lumither 8358a8ae33 feat(backend) update (unfinished) 2025-02-22 00:00:13 -08:00
lumither 84ec754c65 feat(backend) update (unfinished) 2025-02-19 16:24:00 -08:00
lumither bd90cd94d1 fix(backend) fix import error 2025-02-17 16:02:16 -08:00
lumither 5da9ca3ae1 feat(backend) move module scheduler 2025-02-17 15:58:11 -08:00
71 changed files with 2126 additions and 105 deletions

No files matched your search

+3
View File
@@ -8,3 +8,6 @@
# exclude post repo
/blog_posts_git
# rustc ICE dump
rustc-ice-*.txt
+2 -9
View File
@@ -3,14 +3,7 @@ ifeq ($(ARCH), arm64)
ARCH := aarch64
endif
all: release
release:
docker compose -f prod.compose.yaml build --build-arg ARCH=$(ARCH)
dev:
docker-compose -f dev.compose.yaml up --build
.PHONY: up
clean:
docker-compose -f prod.compose.yaml -f dev.compose.yaml rm -fsv
echo "todo"
+1 -7
View File
@@ -7,10 +7,4 @@ content (blog posts etc.) repo.
Powered by Next.js and Rust, built with passion and love
> [!NOTE]
> This system is not production ready, and contains bugs, (security risk potentially).
> But have fun on it =)
# Todos
- [ ] the todo list lol
- [ ] and much more...
> This is an unstable branch, checkout `release` for the (guaranteed) working version
+1 -1
View File
@@ -1,5 +1,5 @@
[workspace]
resolver = "2"
resolver = "3"
members = [
"server",
"macros",
+1 -1
View File
@@ -1,7 +1,7 @@
[package]
name = "macros"
version = "0.1.0"
edition = "2021"
edition = "2024"
[dependencies]
tracing = "0.1.40"
+6
View File
@@ -0,0 +1,6 @@
#[macro_export]
macro_rules! dev_consume {
($($var:ident),+) => {
$(let _ = $var;)*
};
}
+24
View File
@@ -0,0 +1,24 @@
/// Includes a UTF-8 encoded file as a string
///
/// The file is located relative to `CARGO_MANIFEST_DIR`.
///
/// # Example
/// ```rust
/// use macros::include_str_from_crate_root;
///
/// let str: &str = include_str_from_crate_root!("/query/scheme.sql");
///
// todo: impl
// const PATH_PREFIX: &str = "/src";
// let str: &str = include_str_from_crate_root!("{}/query/scheme.sql", PATH_PREFIX);
/// ```
#[macro_export]
macro_rules! include_str_from_crate_root {
// ($path:literal $(, $($arg:tt)*)?) => {
// ($path:expr, $($args:tt)*) => {
// include_str!(concat!(env!("CARGO_MANIFEST_DIR"), format_args!($path, $($args)*)))
// };
($path:expr) => {
include_str!(concat!(env!("CARGO_MANIFEST_DIR"), $path))
};
}
+2
View File
@@ -1 +1,3 @@
mod dev;
mod include_str;
mod log;
+1 -1
View File
@@ -44,7 +44,7 @@ mod tests {
use std::io;
use tracing::Level;
use tracing_subscriber::prelude::*;
use tracing_subscriber::{fmt, Registry};
use tracing_subscriber::{Registry, fmt};
let format = tracing_subscriber::fmt::format()
.with_level(true)
+2 -2
View File
@@ -1,14 +1,14 @@
[package]
name = "markdown"
version = "0.1.0"
edition = "2021"
edition = "2024"
[dependencies]
uuid = { version = "1.10.0", features = ["v4", "fast-rng", "serde"] }
serde = { version = "1.0.209", features = ["derive"] }
serde_json = "1.0.128"
regex = "1.10.4"
yaml-rust2 = "0.9.0"
yaml-rust2 = "0.10.0"
sha2 = "0.10.8"
base64 = "0.22.1"
thiserror = "2.0.3"
-6
View File
@@ -2,12 +2,6 @@ use thiserror::Error;
#[derive(Error, Debug)]
pub enum Error {
#[error("missing front-matter field: id")]
MissingID,
#[error("invalid file id: {}", msg)]
InvalidID { msg: String },
#[error("invalid front-matter: {}", msg)]
InvalidFrontMatter { msg: String },
+19 -30
View File
@@ -1,23 +1,20 @@
mod error;
pub mod error;
mod utils;
use std::fs;
use std::path::Path;
use std::str::FromStr;
use std::{fs, path::Path, str::FromStr};
use crate::error::Error;
use crate::utils::front_matter::{parse_front_matter, split_md_front_matter};
use crate::{
error::Error,
utils::front_matter::{parse_front_matter, split_md_front_matter},
};
use base64::{engine::general_purpose, Engine};
use base64::{Engine, engine::general_purpose};
use serde::Serialize;
use serde_json::Value;
use sha2::{Digest, Sha256};
use uuid::Uuid;
#[derive(Default, Debug, Serialize)]
#[derive(Default, Debug, Serialize, Clone)]
pub struct MdFile {
pub file_id: Uuid,
pub filename: String,
pub meta: Value,
pub content: String,
pub hash: String,
@@ -29,30 +26,22 @@ impl MdFile {
Ok(file) => file,
Err(e) => return Err(Error::FSError { msg: e.to_string() }),
};
let raw_file_bytes = raw_file.as_bytes().to_vec();
let (front_matter_str, content) = split_md_front_matter(raw_file);
Self::from_str(&raw_file)
}
}
impl FromStr for MdFile {
type Err = Error;
fn from_str(s: &str) -> Result<Self, Self::Err> {
let bytes = s.as_bytes().to_vec();
let (front_matter_str, content) = split_md_front_matter(s);
let front_matter = match parse_front_matter(&front_matter_str) {
Ok(front_matter) => front_matter,
Err(e) => return Err(Error::InvalidFrontMatter { msg: e.to_string() }),
};
let hash = calculate_hash_base64(&raw_file_bytes);
let file_id = match front_matter["id"].as_str() {
Some(id) => match Uuid::from_str(id) {
Ok(id) => id,
Err(e) => return Err(Error::InvalidID { msg: e.to_string() }),
},
None => return Err(Error::MissingID),
};
let file_name = Path::new(file_path)
.file_name()
.unwrap()
.to_str()
.unwrap()
.to_string();
let hash = calculate_hash_base64(&bytes);
Ok(MdFile {
file_id,
filename: file_name,
meta: front_matter,
content,
hash,
+9 -8
View File
@@ -1,17 +1,19 @@
use std::error::Error;
use crate::utils::yaml::yaml_to_serde;
use regex::Regex;
use serde_json::Value;
use std::error::Error;
use yaml_rust2::YamlLoader;
pub fn split_md_front_matter(raw_file: String) -> (String, String) {
pub fn split_md_front_matter(raw_file: &str) -> (String, String) {
let re = Regex::new(r"(?s)^(---\n.*?\n---\n)(.*)").unwrap();
if let Some(captures) = re.captures(raw_file.as_str()) {
if let Some(captures) = re.captures(raw_file) {
let front_matter = captures.get(1).map_or("", |m| m.as_str()).to_string();
let main_content = captures.get(2).map_or("", |m| m.as_str()).to_string();
(front_matter, main_content)
} else {
(String::new(), raw_file)
(String::new(), raw_file.to_string())
}
}
@@ -43,8 +45,7 @@ tags: [ "test1", "test2" ]
test body
"##
.to_string(),
"##,
);
assert_eq!(
front_matter,
@@ -64,7 +65,7 @@ tags: [ "test1", "test2" ]
#[test]
fn test_front_matter_parser() {
dbg!(parse_front_matter(
parse_front_matter(
r##"---
title: test_title
date: 0000-00-00+0000
@@ -75,7 +76,7 @@ summary: summary
tags: [ "test1", "test2" ]
---
"##,
))
)
.expect("Failed to parse front matter");
}
}
+17 -3
View File
@@ -1,14 +1,28 @@
[package]
name = "server"
version = "0.1.0"
edition = "2021"
edition = "2024"
[dependencies]
macros = { path = "../macros" }
markdown = { path = "../markdown" }
tracing = "0.1.41"
axum = { version = "0.8.1", features = ["macros"] }
tracing = { version = "0.1.41", features = ["log"] }
git2 = "0.20.0"
chrono = { version = "0.4.39", features = ["serde"] }
cron = "0.15.0"
sqlx = { version = "0.8.3", features = ["runtime-tokio", "tls-native-tls", "postgres", "chrono", "uuid"] }
serde_json = "1.0.133"
serde = { version = "1.0.215", features = ["derive"] }
tracing-subscriber = "0.3.18"
tracing-appender = "0.2.3"
dirs = "6.0.0"
thiserror = "2.0.11"
tokio = { version = "1.42.0", features = ["rt", "rt-multi-thread", "macros", "sync"] }
dotenv = "0.15.0"
uuid = { version = "1.13.2", features = ["v4"] }
futures = "0.3.31"
[build-dependencies]
git2 = "0.20.0"
+3 -2
View File
@@ -9,7 +9,8 @@ fn main() {
.peel_to_commit()
.expect("failed to get HEAD commit");
let hash = commit.id().to_string();
let utc_date = Utc::now().date_naive().format("%Y%m%d").to_string();
let build_id = format!("{}@{}", utc_date, &hash[..8]);
let utc_timestamp = Utc::now().date_naive().format("%Y%m%d").to_string();
let pkg_version = env!("CARGO_PKG_VERSION");
let build_id = format!("{}:{}@{}", pkg_version, &hash[..8], utc_timestamp);
println!("cargo:rustc-env=BUILD_ID={}", build_id);
}
-32
View File
@@ -1,32 +0,0 @@
CREATE TABLE tag
(
id SERIAL PRIMARY KEY,
name VARCHAR(20) UNIQUE NOT NULL,
description TEXT
);
CREATE TABLE category
(
id SERIAL PRIMARY KEY,
name VARCHAR(20) UNIQUE NOT NULL,
description TEXT
);
CREATE TABLE post
(
id SERIAL PRIMARY KEY,
identifier VARCHAR(50) UNIQUE NOT NULL,
title VARCHAR(255) NOT NULL,
subtitle VARCHAR(255) NOT NULL,
published_date DATE NOT NULL,
last_update DATE,
content TEXT NOT NULL,
category SERIAL REFERENCES category (id) NOT NULL
);
CREATE TABLE post_tag
(
post SERIAL REFERENCES post (id),
tag SERIAL REFERENCES tag (id)
)
+2
View File
@@ -1,3 +1,5 @@
// todo: rebuild with static OnceCell
// Database related
pub mod database {
pub mod var_name {
+2 -1
View File
@@ -1,2 +1,3 @@
pub const DEFAULT_TIME_FORMAT: &str = "%Y-%m-%dT%H:%M:%S%z";
pub const DEFAULT_DATE_FORMAT: &str = "%Y-%m-%d%zT%H%M%S";
pub const DEFAULT_NAIVE_DATE_FORMAT: &str = "%Y-%m-%d";
+42
View File
@@ -0,0 +1,42 @@
use std::env;
use std::error::Error;
use std::str::FromStr;
use crate::constants::config::database::{default_value, var_name};
use sqlx::postgres::{PgConnectOptions, PgPoolOptions};
use sqlx::{ConnectOptions, Pool, Postgres};
use tracing::info;
pub async fn init() -> Result<Pool<Postgres>, Box<dyn Error>> {
let db_connect_option = PgConnectOptions::from_str(&parse_db_uri())
.unwrap()
.disable_statement_logging();
Ok(PgPoolOptions::new()
.max_connections(1) // todo: temporary solution, to be fixed
.connect_with(db_connect_option)
.await?)
}
fn parse_db_uri() -> String {
env::var(var_name::URI).unwrap_or_else(|_| {
info!(
"`{}` env var is not set, \
using `{}`, `{}`, `{}`, `{}` and `{}`",
var_name::URI,
var_name::USERNAME,
var_name::PASSWORD,
var_name::HOST,
var_name::PORT,
var_name::DBNAME
);
let postgres_user = env::var(var_name::USERNAME).unwrap_or(default_value::USERNAME.into());
let postgres_password =
env::var(var_name::PASSWORD).unwrap_or(default_value::PASSWORD.into());
let db_host = env::var(var_name::HOST).unwrap_or(default_value::HOST.into());
let db_port = env::var(var_name::PORT).unwrap_or(default_value::PORT.into());
let postgres_db = env::var(var_name::DBNAME).unwrap_or(default_value::DBNAME.into());
let db_type = default_value::DATABASE_TYPE;
format!("{db_type}://{postgres_user}:{postgres_password}@{db_host}:{db_port}/{postgres_db}")
})
}
+107
View File
@@ -0,0 +1,107 @@
use std::path::Path;
use std::{env, io};
use crate::constants::config::general::{default_value, var_name};
use chrono::Local;
use dirs::home_dir;
use tracing::info;
use tracing_appender::{non_blocking, non_blocking::WorkerGuard, rolling};
use tracing_subscriber::{
Layer, Registry, filter::LevelFilter, fmt, fmt::MakeWriter, fmt::format::Writer,
fmt::time::FormatTime, layer::SubscriberExt,
};
struct LocalTimer;
impl FormatTime for LocalTimer {
fn format_time(&self, w: &mut Writer<'_>) -> std::fmt::Result {
write!(
w,
"{}",
Local::now().format(crate::constants::time::DEFAULT_TIME_FORMAT)
)
}
}
type _LogWriter = dyn for<'writer> MakeWriter<'writer, Writer = dyn io::Write + 'static> + 'static;
pub fn init() -> (WorkerGuard, WorkerGuard) {
let log_path = match env::var(var_name::LOG_DIR) {
Ok(path) => Path::new(&path).to_path_buf(),
Err(_) => {
let path = if let Ok(default_log_path) = env::var(var_name::WORK_DIR) {
expand_home_path(default_log_path)
} else {
let fallback_path = expand_home_path(default_value::WORK_DIR.to_string());
println!(
"[warn] unset env var `{}` and `{}`, trying to log on fallback path: {}/log",
var_name::LOG_DIR,
var_name::WORK_DIR,
fallback_path
);
fallback_path
};
Path::new(&path).join("log")
}
};
let (info_log_writer, info_guard) = non_blocking(rolling::never(&log_path, "info.log"));
let (err_log_writer, error_guard) = non_blocking(rolling::never(&log_path, "error.log"));
let info_format = tracing_subscriber::fmt::format()
.with_level(true)
.with_target(true)
.with_source_location(true)
.with_ansi(false);
let err_format = tracing_subscriber::fmt::format()
.with_level(true)
.with_target(true)
.with_source_location(true)
.with_ansi(false);
let stdout_format = tracing_subscriber::fmt::format()
.with_level(true)
.with_target(true)
.with_source_location(true)
.with_timer(LocalTimer);
let subscribers = Registry::default()
.with(
fmt::Layer::default()
.with_writer(info_log_writer)
.event_format(info_format),
)
.with(
fmt::Layer::default()
.with_writer(err_log_writer)
.event_format(err_format)
.with_filter(LevelFilter::ERROR),
)
.with(
fmt::Layer::default()
.with_writer(io::stdout)
.with_ansi(true)
.event_format(stdout_format),
);
tracing::subscriber::set_global_default(subscribers)
.expect("[fatal] failed to setup logging system");
info!("logging to dir: {}", log_path.to_str().unwrap());
(info_guard, error_guard)
}
fn expand_home_path(path: String) -> String {
if path.starts_with('~') {
let home_path = home_dir()
.unwrap_or_else(|| panic!("[fatal] unable to locate home dir when parsing env `{}`, try to use absolute path instead", var_name::WORK_DIR));
home_path
.join(path.strip_prefix("~/").unwrap())
.into_os_string()
.into_string()
.unwrap()
} else {
path
}
}
+14
View File
@@ -0,0 +1,14 @@
use git2::opts::{set_server_connect_timeout_in_milliseconds, set_server_timeout_in_milliseconds};
use tracing::warn;
pub fn init() {
// git2 option
unsafe {
if let Err(e) = set_server_connect_timeout_in_milliseconds(10_000) {
warn!("failed to set git2 server connect timeout: {}", e);
}
if let Err(e) = set_server_timeout_in_milliseconds(60_000) {
warn!("failed to set git2 server timeout: {}", e);
}
}
}
+50
View File
@@ -0,0 +1,50 @@
mod database;
pub mod logger;
mod misc;
use std::{env, panic};
use crate::modules::{ModuleTree, blog::Blog};
use macros::error_panic;
use tracing::info;
pub fn load_env() {
if let Some(env_file) = env::args().nth(1) {
match dotenv::from_filename(&env_file) {
Ok(_) => {
println!("[info] env loaded from {}, starting up", env_file);
}
Err(e) => {
panic!("[fatal] failed to read {}: {}", env_file, e);
}
}
} else {
println!("[info] no .env file referred, starting up")
}
}
pub async fn start() {
misc::init();
// connect database
let db_pool = match database::init().await {
Ok(pool) => {
info!("database connection pool initialized");
pool
}
Err(e) => {
error_panic!("database connection pool failed to initialize: {}", e);
}
};
// // schema init
// schema::init(&db_pool).await;
ModuleTree::new()
.add_module(Blog {
db_handler: db_pool,
})
.blocking_serve()
.await
}
+16 -1
View File
@@ -1,7 +1,22 @@
#[allow(dead_code)]
mod constants;
mod scheduler;
mod init;
mod modules;
mod utils;
use crate::init::{logger, start};
const BUILD_ID: &str = env!("BUILD_ID");
fn main() {
#[tokio::main]
async fn main() {
println!("{}", BUILD_ID);
init::load_env();
let _guards = logger::init();
start().await;
}
@@ -0,0 +1,25 @@
use crate::constants::blog::PAGE_SIZE;
use axum::{Json, extract::State};
use serde_json::{Value, json};
use sqlx::{PgPool, Row, query};
use tracing::error;
const GET_PAGE_COUNT_QUERY: &str = include_str!("post_page_count.sql");
pub async fn handler(State(db_connection): State<PgPool>) -> Result<Json<Value>, Json<Value>> {
match query(GET_PAGE_COUNT_QUERY)
.bind(PAGE_SIZE as i64)
.fetch_one(&db_connection)
.await
{
Ok(res) => {
let res: i32 = res.get("count");
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()})))
}
}
}
@@ -0,0 +1,2 @@
SELECT cast(ceil(count(*) / cast($1 AS FLOAT)) AS INTEGER) AS count
FROM posts;
@@ -0,0 +1 @@
SELECT
Whitespace-only changes.
@@ -0,0 +1,35 @@
use macros::dev_consume;
use axum::{
Json,
extract::{Path, State},
response::IntoResponse,
};
use serde_json::json;
use sqlx::PgPool;
use uuid::Uuid;
const GET_POST_BY_UUID: &str = include_str!("get_post_by_uuid.sql");
const GET_POST_BY_STR_ID: &str = include_str!("get_post_by_str_id.sql");
pub async fn handler(
State(db_connection): State<PgPool>,
Path(identifier): Path<String>,
) -> impl IntoResponse {
if let Ok(id) = identifier.parse::<Uuid>() {
select_post_by_uuid(&db_connection, id).await
} else {
select_post_by_str_id(&db_connection, &identifier).await
};
Json::from(json!({"res": "ok"})).into_response()
}
async fn select_post_by_uuid(db: &PgPool, id: Uuid) {
dev_consume!(db, id);
todo!()
}
async fn select_post_by_str_id(db: &PgPool, id: &str) {
dev_consume!(db, id);
todo!()
}
@@ -0,0 +1,130 @@
use std::error::Error;
use crate::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>,
}
#[axum::debug_handler]
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 = Vec::new();
let posts: Vec<_> = post_list
.into_iter()
.map(|post| async { analyze_post(&db_connection, post).await })
.collect();
let posts = join_all(posts).await;
for post in posts {
match post {
Ok(p) => res.push(p),
Err(e) => {
return Err((
StatusCode::INTERNAL_SERVER_ERROR,
Json::from(json!(("error", e.to_string()))),
))
}
}
}
// res.insert("posts".to_string(), Value::from(posts));
Ok(Json::from(Value::from(res)))
}
async fn analyze_post(db_connection: &PgPool, post: PgRow) -> Result<Value, Box<dyn Error + Send>> {
let post_id: Uuid = post.get("post_id");
let tags = fetch_post_tags(db_connection, post_id).await;
let meta = fetch_post_meta(db_connection, post_id).await.unwrap();
let header_img = meta.get("header_img");
let last_update = meta.get("last_update");
Ok(json!({
"post_id": post_id.to_string(),
"title": post.get::<String, _>("title"),
"category": post.get::<String, _>("category"),
"header_img": header_img,
"sub_title": post.get::<String, _>("sub_title"),
"summary": post.get::<String, _>("summary"),
"last_update": last_update,
"first_update": post.get::<DateTime<Utc>, _>("first_update"),
"tags": tags,
}))
}
async fn fetch_post_meta(db_connection: &PgPool, post_id: Uuid) -> Option<Value> {
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) => Some(value.get::<Value, _>("meta")),
Err(e) => {
error!(
"Database error on Reading `{}` from table `{}`: <{}>: {}",
"post meta data",
"meta",
post_id.to_string(),
e.to_string()
);
None
}
}
}
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![],
}
}
@@ -0,0 +1,6 @@
pub mod get_page_count;
// todo: remove exception tag
#[allow(dead_code)]
pub mod get_post;
// pub mod get_post_list;
@@ -0,0 +1,208 @@
use crate::modules::blog::components::check_update::{
changes::{Change, Create, Delete, Move, Update},
error::Error,
utils::{parse_naive_date_str, parse_post_identifier},
};
use futures::{
future::{join_all, BoxFuture},
FutureExt,
};
use sqlx::{query, PgPool, Postgres, Row, Transaction};
use tracing::error;
const DELETE_POST_BY_ID: &str = include_str!("sql/delete_post_by_id.sql");
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");
pub async fn apply_deltas(db_conn: &PgPool, changes: Vec<Change>) {
let handlers = changes
.into_iter()
.map(|change| match change {
Change::Delete(del) => handle_delete(db_conn, del).boxed(),
Change::Move(mv) => handle_move(db_conn, mv).boxed(),
Change::Update(update) => handle_update(db_conn, update).boxed(),
Change::Create(create) => handle_create(db_conn, create).boxed(),
})
.collect::<Vec<BoxFuture<'_, ()>>>();
join_all(handlers).await;
}
async fn handle_delete(db: &PgPool, del: Delete) {
if let Err(e) = query(DELETE_POST_BY_ID).bind(del.uuid).execute(db).await {
error!(
"failed to delete post: {}, {}",
del.uuid.to_string(),
e.to_string()
)
}
}
async fn handle_move(db: &PgPool, mv: Move) {
handle_delete(
db,
Delete {
uuid: mv.uuid,
path: mv.from,
},
)
.await;
handle_create(
db,
Create {
uuid: mv.uuid,
path: mv.to,
payload: mv.payload,
},
)
.await;
}
async fn handle_update(db: &PgPool, update: Update) {
handle_delete(
db,
Delete {
uuid: update.uuid,
path: update.path.clone(),
},
)
.await;
handle_create(
db,
Create {
uuid: update.uuid,
path: update.path,
payload: update.payload,
},
)
.await;
}
async fn handle_create(db: &PgPool, create: Create) {
let mut tx = db.begin().await.unwrap();
let identifier = parse_post_identifier(&create.path).unwrap();
let tags_json = create.payload.meta["tags"].clone();
let tags: Vec<_> = if let Some(tags) = tags_json.as_array() {
tags.iter().filter_map(|tag| tag.as_str()).collect()
} else {
vec![]
};
let mut tag_ids = vec![];
for tag in tags {
match get_or_create_tag_id(&mut tx, tag).await {
Ok(tid) => tag_ids.push(tid),
Err(e) => {
error!("failed to fetch or create tag ({}): {}", tag, e);
return;
}
}
}
let category_json = create.payload.meta["category"].clone();
let category = category_json.as_str().unwrap_or("Uncategorized");
let category_id = match get_or_create_category_id(&mut tx, category).await {
Ok(id) => id,
Err(e) => {
error!(
"failed to fetch category id for post {}: {}",
&identifier, e
);
return;
}
};
let meta = create.payload.meta;
let post_id = create.uuid;
let post_identifier = &identifier;
let post_title = meta
.get("title")
.map(|t| t.as_str().unwrap())
.unwrap_or("Untitled");
let post_subtitle = meta
.get("subtitle")
.map(|t| t.as_str().unwrap())
.unwrap_or("");
let post_date_created = if let Some(Some(date)) = meta
.get("date")
.map(|d| parse_naive_date_str(d.as_str().unwrap_or("")))
{
date
} else {
error!("failed to parse `date` for post: {}", post_identifier);
return;
};
let post_date_updated = meta
.get("update")
.map(|d| parse_naive_date_str(d.as_str().unwrap_or("")))
.unwrap_or(None);
let post_status = meta.get("status").map(|v| v.as_str()).unwrap_or(None);
let post_content = create.payload.content;
if let Err(e) = query(CREATE_POST)
.bind(post_id)
.bind(post_identifier)
.bind(post_title)
.bind(post_subtitle)
.bind(post_date_created)
.bind(post_date_updated)
.bind(post_status)
.bind(post_content)
.bind(category_id)
.bind(tag_ids)
.execute(&mut *tx)
.await
{
error!("failed to insert post {}: {}", &identifier, e);
return;
}
tx.commit().await.unwrap();
}
async fn get_or_create_tag_id(
db: &mut Transaction<'_, Postgres>,
tag_name: &str,
) -> Result<i32, Error> {
get_create_id(db, tag_name, GET_TAG_ID, CREATE_RETURN_TAG_ID).await
}
async fn get_or_create_category_id(
db: &mut Transaction<'_, Postgres>,
category_name: &str,
) -> Result<i32, Error> {
get_create_id(
db,
category_name,
GET_CATEGORY_ID,
CREATE_RETURN_CATEGORY_ID,
)
.await
}
async fn get_create_id(
db: &mut Transaction<'_, Postgres>,
field_name: &str,
fetch_query: &str,
create_fetch_query: &str,
) -> Result<i32, Error> {
if let Some(id) = query(fetch_query)
.bind(field_name)
.fetch_optional(&mut **db)
.await?
{
Ok(id.get("id"))
} else {
let id = query(create_fetch_query)
.bind(field_name)
.fetch_one(&mut **db)
.await?
.get("id");
Ok(id)
}
}
@@ -0,0 +1,68 @@
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 path: PathBuf,
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,
}
}
}
@@ -0,0 +1,18 @@
use markdown::error::Error as MarkdownError;
use thiserror::Error;
#[derive(Error, Debug)]
pub enum Error {
#[error("markdown file error")]
MdFile(#[from] MarkdownError),
#[error("db error")]
Db(#[from] sqlx::Error),
#[error("git error")]
Git2(#[from] git2::Error),
#[error("git remote url mismatch")]
GitUrlMismatch,
}
@@ -0,0 +1,103 @@
use std::{
env,
path::{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,
},
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;
#[derive(Debug)]
struct Args {
db_connection: PgPool,
git_work_dir: PathBuf,
git_url: String,
git_remote_name: String,
git_remote_branch: String,
}
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,
Err(e) => {
panic_with_log!(
Level::ERROR,
"failed to read `{}`: {}",
var_name::BLOG_GIT_URL,
e
)
}
};
let git_remote_name = env::var(var_name::BLOG_GIT_REMOTE_NAME)
.unwrap_or(default_value::BLOG_GIT_REMOTE_NAME.to_string());
let git_remote_branch = env::var(var_name::BLOG_GIT_REMOTE_BRANCH)
.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(workflow(arc_args.clone()))
}
async fn workflow(args: Arc<Args>) {
let deltas = match fetch_deltas(
&args.git_url,
&args.git_remote_name,
&args.git_remote_branch,
&args.git_work_dir,
)
.await
{
Ok(updates) => updates,
Err(e) => {
error!("delta parsing error: {}", e);
return;
}
};
if deltas.is_empty() {
trace!("blog git already up-to-date");
return;
}
apply_deltas(&args.db_connection, deltas).await;
}
@@ -0,0 +1,171 @@
use std::{collections::HashMap, fs::remove_dir_all, path::PathBuf};
use crate::{
modules::blog::components::check_update::{
changes::{Change, Create, CreateDelete2UpdateSlot, Delete, Update},
error::Error,
},
utils::{git, git::FileDelta},
};
use markdown::MdFile;
use git2::{Delta, Repository};
use tracing::{error, warn};
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 = match repo.find_remote(git_remote_name) {
Ok(remote) => remote,
Err(e) => {
remove_dir_all(git_work_dir).unwrap();
warn!(
"cannot find git remote ({}), reinitializing: {}",
git_remote_name, e
);
return Err(e.into());
}
};
if remote.url().is_some_and(|url| url != git_url) {
remove_dir_all(git_work_dir).unwrap();
warn!(
"git remote url mismatch (expected: {}, actual: {}), reinitializing",
git_url,
remote.url().unwrap()
);
return Err(Error::GitUrlMismatch);
}
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() {
// todo: limit scanning scope to post only OR extract merging logic to external component
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,
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::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) => 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();
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
}
@@ -0,0 +1,7 @@
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);
@@ -0,0 +1,3 @@
INSERT INTO categories(name)
VALUES ($1)
RETURNING id;
@@ -0,0 +1,3 @@
INSERT INTO tags(name)
VALUES ($1)
RETURNING id;
@@ -0,0 +1,3 @@
DELETE
FROM posts
WHERE id = $1;
@@ -0,0 +1,3 @@
SELECT id
FROM categories
WHERE name = $1;
@@ -0,0 +1,3 @@
SELECT id
FROM tags
WHERE name = $1;
@@ -0,0 +1,58 @@
use std::path::Path;
use crate::constants::{
config::general::{default_value, var_name},
time::DEFAULT_NAIVE_DATE_FORMAT,
};
use chrono::NaiveDate;
use dirs::home_dir;
use tracing::warn;
pub fn expand_path(path: String) -> String {
if path.starts_with('~') {
home_dir()
.map(|dir| {
dir.join(path.strip_prefix("~/").unwrap())
.into_os_string()
.into_string()
.unwrap()
})
.unwrap_or_else(|| {
warn!(
"failed to parse env var `{}`, using default value `{}`",
var_name::WORK_DIR,
default_value::WORK_DIR
);
default_value::WORK_DIR.to_string()
})
} else {
path
}
}
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())
} else {
None
}
}
pub fn parse_naive_date_str(date_str: &str) -> Option<NaiveDate> {
NaiveDate::parse_from_str(date_str, DEFAULT_NAIVE_DATE_FORMAT).ok()
}
#[cfg(test)]
mod tests {
use crate::modules::blog::components::check_update::utils::parse_naive_date_str;
use chrono::NaiveDate;
#[test]
fn test_parse_naive_date() {
assert_eq!(
NaiveDate::from_ymd_opt(2015, 1, 1),
parse_naive_date_str("2015-1-01")
);
}
}
@@ -0,0 +1,3 @@
pub mod check_update;
mod consistency_guard;
mod rag;
@@ -0,0 +1 @@
+55
View File
@@ -0,0 +1,55 @@
mod api;
mod components;
use crate::modules::blog::components::check_update;
use crate::scheduler::tasks::async_basic::AsyncBasic;
use crate::{
modules::{
Module,
blog::{
api::{
get_page_count,
get_post,
// get_post_list::get_post_list,
},
// cron::check_update,
},
},
scheduler::tasks::CronTask,
};
use axum::{Router, routing::get};
use sqlx::PgPool;
const MOUNT_POINT: &str = "/blog";
pub struct Blog {
pub db_handler: PgPool,
}
impl Module for Blog {
fn get_server_router(&self) -> Router {
let db_handler = self.db_handler.clone();
// todo: move resource init to new func
// todo: remove debug comment
Router::new()
// .route("/get_post_list", get(get_post_list))
.route("/get_page_count", get(get_page_count::handler))
.route("/get_post/{identifier}", get(get_post::handler))
.with_state(db_handler)
}
fn get_mount_point(&self) -> &'static str {
MOUNT_POINT
}
fn get_cron_tasks(&self) -> Vec<(&str, Box<dyn CronTask>)> {
let db_handler = self.db_handler.clone();
vec![(
"Blog Check Update",
AsyncBasic::new(check_update::task(db_handler.clone()), "*/5 * * * * *")
.unwrap()
.to_task(),
)]
}
}
@@ -0,0 +1,65 @@
CREATE TABLE tags
(
id SERIAL PRIMARY KEY,
name VARCHAR(20) UNIQUE NOT NULL,
description TEXT DEFAULT ''
);
CREATE TABLE categories
(
id SERIAL PRIMARY KEY,
name VARCHAR(20) UNIQUE NOT NULL,
description TEXT DEFAULT ''
);
CREATE TABLE post_status
(
id SERIAL PRIMARY KEY,
name VARCHAR(20) UNIQUE NOT NULL
);
INSERT INTO post_status (name)
VALUES ('draft'),
('published'),
('planned'),
('hidden');
CREATE TABLE posts
(
id UUID PRIMARY KEY,
identifier VARCHAR(100) UNIQUE NOT NULL,
title VARCHAR(255) NOT NULL,
subtitle VARCHAR(255) NOT NULL,
date_created DATE NOT NULL,
date_updated DATE,
status INTEGER REFERENCES post_status (id) DEFAULT 1 NOT NULL,
content TEXT NOT NULL,
category INTEGER REFERENCES categories (id) NOT NULL
);
CREATE TABLE post_tag
(
post UUID REFERENCES posts (id) ON DELETE CASCADE,
tag INTEGER REFERENCES tags (id) ON DELETE CASCADE,
CONSTRAINT post_tag_pk PRIMARY KEY (post, tag)
);
CREATE TABLE metadata
(
pid UUID REFERENCES posts (id) ON DELETE CASCADE NOT NULL,
meta_key VARCHAR(50) NOT NULL,
meta_value TEXT NOT NULL,
CONSTRAINT metadata_pk PRIMARY KEY (pid, meta_key)
);
-- index
CREATE INDEX idx_tags_name ON tags (name);
CREATE INDEX idx_categories_name ON categories (name);
CREATE INDEX idx_posts_identifier ON posts (identifier);
CREATE INDEX idx_posts_status ON posts (status);
CREATE INDEX idx_posts_category ON posts (category);
CREATE INDEX idx_post_tag_post ON post_tag (post);
CREATE INDEX idx_post_tag_tag ON post_tag (tag);
CREATE INDEX idx_metadata_pid ON metadata (pid);
+82
View File
@@ -0,0 +1,82 @@
use std::{env, num::ParseIntError};
use crate::{
constants::config::server::{default_value, var_name},
scheduler::{Scheduler, tasks::CronTask},
};
use macros::error_panic;
use axum::Router;
use tokio::net::TcpListener;
use tracing::{info, warn};
pub mod blog;
// todo: merge `get_server_router` & `get_mount_point`, make Option
pub trait Module {
fn get_server_router(&self) -> Router;
fn get_mount_point(&self) -> &'static str;
fn get_cron_tasks(&self) -> Vec<(&str, Box<dyn CronTask>)>;
}
pub struct ModuleTree {
pub modules: Vec<Box<dyn Module>>,
}
impl ModuleTree {
pub fn new() -> Self {
ModuleTree { modules: vec![] }
}
pub fn add_module(&mut self, module: impl Module + 'static) -> &mut Self {
self.modules.push(Box::new(module));
self
}
pub async fn blocking_serve(&self) {
let mut scheduler = Scheduler::new();
let mut app = Router::new();
let port: u32 = match env::var(var_name::PORT) {
Ok(value) => value.parse().unwrap_or_else(|e: ParseIntError| {
warn!(
"failed to parse `{}`, using default port: {}",
var_name::PORT,
e.to_string()
);
default_value::PORT
}),
Err(_) => default_value::PORT,
};
let listener = match TcpListener::bind(format!("{}:{}", default_value::HOST, port)).await {
Ok(listener) => {
info!("server starting on {}:{}", default_value::HOST, port);
listener
}
Err(e) => {
error_panic!(
"failed to start server on {}:{}: {}",
default_value::HOST,
port,
e.to_string()
);
}
};
for module in &self.modules {
let _ = scheduler.insert_list(module.get_cron_tasks());
app = app.nest(module.get_mount_point(), module.get_server_router());
}
tokio::spawn(async move {
scheduler.start().await;
});
axum::serve(listener, app).await.unwrap_or_else(|e| {
error_panic!("failed to start axum: {}", e.to_string());
});
}
}
+49
View File
@@ -0,0 +1,49 @@
use std::{
error::Error,
fmt::{Debug, Display, Formatter},
sync::PoisonError,
};
use crate::scheduler::error::SchedulerError::MutexPoison;
// todo: refactor with thiserror
#[allow(dead_code)]
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 {}
+54
View File
@@ -0,0 +1,54 @@
use std::{collections::HashMap, sync::Arc};
use crate::scheduler::{error::SchedulerError, tasks::CronTask, utils::execute};
use futures::future::join_all;
use tracing::warn;
pub mod error;
pub mod task_func;
pub mod tasks;
mod utils;
pub type TaskPool = HashMap<String, Arc<Box<dyn CronTask>>>;
#[derive(Default)]
pub struct Scheduler {
task_pool: TaskPool,
}
impl Scheduler {
/// Start all tasks in the scheduler
pub async fn start(&self) {
let fut = self
.task_pool
.clone()
.into_iter()
.map(|(name, task)| execute(name, task))
.collect::<Vec<_>>();
// self.runner.run_all().await
join_all(fut).await;
}
pub fn insert(&mut self, task: (&str, Box<dyn CronTask>)) -> Result<&Self, SchedulerError> {
let (k, v) = task;
if self.task_pool.insert(k.into(), v.into()).is_some() {
warn!("task list duplicated key detected: {}", k);
}
Ok(self)
}
pub fn insert_list(
&mut self,
tasks: Vec<(&str, Box<dyn CronTask>)>,
) -> Result<&Self, SchedulerError> {
for task in tasks {
self.insert(task)?;
}
Ok(self)
}
pub fn new() -> Self {
Self::default()
}
}
+15
View File
@@ -0,0 +1,15 @@
use std::pin::Pin;
pub trait TaskFunc: Fn() + Send + Sync + 'static {}
impl<T> TaskFunc for T where T: Fn() + Send + Sync + Clone + 'static {}
pub trait AsyncTaskFunc:
Fn() -> Pin<Box<dyn Future<Output = ()> + Send>> + Send + Sync + 'static
{
}
impl<T> AsyncTaskFunc for T where
T: Fn() -> Pin<Box<dyn Future<Output = ()> + Send>> + Send + Sync + Clone + 'static
{
}
@@ -0,0 +1,37 @@
use std::{error::Error, pin::Pin, str::FromStr};
use crate::{scheduler::task_func::AsyncTaskFunc, scheduler::tasks::CronTask};
use chrono::{DateTime, Utc};
use cron::Schedule;
pub struct AsyncBasic {
schedule: Schedule,
func: Box<dyn AsyncTaskFunc>,
}
impl AsyncBasic {
pub fn new<F>(func: F, cron_expr: &str) -> Result<Self, Box<dyn Error>>
where
F: AsyncTaskFunc,
{
Ok(AsyncBasic {
schedule: Schedule::from_str(cron_expr)?,
func: Box::new(func),
})
}
pub fn to_task(self) -> Box<dyn CronTask> {
Box::new(self)
}
}
impl CronTask for AsyncBasic {
fn get_next_execution(&self) -> Option<DateTime<Utc>> {
self.schedule.upcoming(Utc).next()
}
fn call(&self) -> Pin<Box<(dyn Future<Output = ()> + Send)>> {
Box::pin((self.func)())
}
}
@@ -0,0 +1,37 @@
use std::{error::Error, pin::Pin, str::FromStr};
use crate::scheduler::{task_func::TaskFunc, tasks::CronTask};
use chrono::{DateTime, Utc};
use cron::Schedule;
pub struct BasicTask {
schedule: Schedule,
func: Box<dyn TaskFunc>,
}
impl BasicTask {
pub fn new<F>(func: F, cron_expr: &str) -> Result<Self, Box<dyn Error>>
where
F: TaskFunc,
{
Ok(BasicTask {
schedule: Schedule::from_str(cron_expr)?,
func: Box::new(func),
})
}
pub fn to_task(self) -> Box<dyn CronTask> {
Box::new(self)
}
}
impl CronTask for BasicTask {
fn get_next_execution(&self) -> Option<DateTime<Utc>> {
self.schedule.upcoming(Utc).next()
}
fn call(&self) -> Pin<Box<dyn Future<Output = ()> + Send + '_>> {
Box::pin(async { (self.func)() })
}
}
+14
View File
@@ -0,0 +1,14 @@
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 {
fn get_next_execution(&self) -> Option<DateTime<Utc>>;
fn call(&self) -> Pin<Box<dyn Future<Output = ()> + Send + '_>>;
}
+32
View File
@@ -0,0 +1,32 @@
use std::{sync::Arc, time::Duration};
use crate::scheduler::{error::SchedulerError, tasks::CronTask};
use chrono::{DateTime, Utc};
use tokio::time::{Sleep, sleep};
pub fn duration_from_now(future: DateTime<Utc>) -> Duration {
future
.signed_duration_since(Utc::now())
.to_std()
.unwrap_or(Duration::ZERO)
}
pub fn sleeper_from_now(future: DateTime<Utc>) -> Sleep {
sleep(duration_from_now(future))
}
#[allow(unused_variables)]
pub async fn execute(name: String, func: Arc<Box<dyn CronTask>>) -> Result<(), SchedulerError> {
// todo: stats
while let Some(next_execution) = func.get_next_execution() {
let sleeper = sleeper_from_now(next_execution);
tokio::pin!(sleeper);
tokio::select! {
_ = sleeper => {
func.call().await
}
}
}
Ok(())
}
+166
View File
@@ -0,0 +1,166 @@
/// 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
///
use git2::{
AnnotatedCommit, AutotagOption, Delta, FetchOptions, Reference, Remote, Repository, build,
};
use std::path::PathBuf;
use tracing::{error, warn};
#[derive(Debug, Eq, PartialEq)]
pub struct FileDelta {
pub old_path: Option<PathBuf>,
pub new_path: Option<PathBuf>,
pub status: Delta,
}
pub fn do_fetch<'a>(
repo: &'a Repository,
refs: &[&str],
remote: &'a mut Remote,
) -> Result<AnnotatedCommit<'a>, git2::Error> {
let mut fetch_option = FetchOptions::new();
fetch_option.download_tags(AutotagOption::All);
remote.fetch(refs, Some(&mut fetch_option), None)?;
let fetch_head = repo.find_reference("FETCH_HEAD")?;
repo.reference_to_annotated_commit(&fetch_head)
}
fn fast_forward(
repo: &Repository,
lb: &mut Reference,
rc: &AnnotatedCommit,
) -> Result<(), git2::Error> {
let name = match lb.name() {
Some(s) => s.to_string(),
None => String::from_utf8_lossy(lb.name_bytes()).to_string(),
};
let msg = format!("fast-forward({}) -> {}", name, rc.id());
lb.set_target(rc.id(), &msg)?;
repo.set_head(&name)?;
repo.checkout_head(Some(build::CheckoutBuilder::default().force()))?;
Ok(())
}
fn normal_merge(
repo: &Repository,
local: &AnnotatedCommit,
remote: &AnnotatedCommit,
) -> Result<(), git2::Error> {
let local_tree = repo.find_commit(local.id())?.tree()?;
let remote_tree = repo.find_commit(remote.id())?.tree()?;
let ancestor = repo
.find_commit(repo.merge_base(local.id(), remote.id())?)?
.tree()?;
let mut idx = repo.merge_trees(&ancestor, &local_tree, &remote_tree, None)?;
if idx.has_conflicts() {
warn!("merge conflicts detected");
repo.checkout_index(Some(&mut idx), None)?;
return Ok(());
}
let result_tree = repo.find_tree(idx.write_tree_to(repo)?)?;
let msg = format!("merge({}) -> {}", remote.id(), local.id());
let sig = repo.signature()?;
let local_commit = repo.find_commit(local.id())?;
let remote_commit = repo.find_commit(remote.id())?;
let _merge_commit = repo.commit(
Some("HEAD"),
&sig,
&sig,
&msg,
&result_tree,
&[&local_commit, &remote_commit],
)?;
repo.checkout_head(None)?;
Ok(())
}
pub fn do_merge<'a>(
repo: &'a Repository,
remote_branch: &str,
fetch_commit: AnnotatedCommit<'a>,
) -> Result<(), git2::Error> {
let analysis = repo.merge_analysis(&[&fetch_commit])?;
let head_commit = repo.reference_to_annotated_commit(&repo.head()?)?;
if analysis.0.is_fast_forward() {
let ref_name = format!("refs/heads/{}", remote_branch);
match repo.find_reference(&ref_name) {
Ok(mut r) => {
fast_forward(repo, &mut r, &fetch_commit)?;
}
Err(_) => {
repo.reference(
&ref_name,
fetch_commit.id(),
true,
&format!("({}) -> {}", remote_branch, fetch_commit.id()),
)?;
repo.set_head(&ref_name)?;
repo.checkout_head(Some(
build::CheckoutBuilder::default()
.allow_conflicts(true)
.conflict_style_merge(true)
.force(),
))?;
}
};
} else if analysis.0.is_normal() {
normal_merge(repo, &head_commit, &fetch_commit)?;
}
Ok(())
}
pub fn calculate_diff(
repo: &Repository,
local: &AnnotatedCommit,
remote: &AnnotatedCommit,
) -> Result<Vec<FileDelta>, git2::Error> {
let local_tree = repo.find_commit(local.id())?.tree()?;
let remote_tree = repo.find_commit(remote.id())?.tree()?;
let diff = repo.diff_tree_to_tree(Some(&local_tree), Some(&remote_tree), None)?;
let mut updated_files: Vec<FileDelta> = Vec::new();
if let Err(e) = diff.foreach(
&mut |d, _| {
updated_files.push(FileDelta {
old_path: d.old_file().path().map(PathBuf::from),
new_path: d.new_file().path().map(PathBuf::from),
status: d.status(),
});
true
},
None,
None,
None,
) {
error!("error at calculating diff: {:?}", e);
}
Ok(updated_files.into_iter().collect())
}
#[allow(dead_code)]
pub fn sync(
remote_name: &str,
remote_branch: &str,
repo: &Repository,
) -> Result<Vec<FileDelta>, git2::Error> {
let mut remote = repo.find_remote(remote_name)?;
let fetch_commit = do_fetch(repo, &[remote_branch], &mut remote)?;
let delta = calculate_diff(
repo,
&repo.reference_to_annotated_commit(&repo.head()?)?,
&fetch_commit,
);
do_merge(repo, remote_branch, fetch_commit)?;
delta
}
+1
View File
@@ -0,0 +1 @@
pub mod git;
+11
View File
@@ -0,0 +1,11 @@
.venv/
.idea/
.DS_store
__pycache__
.cache
content/*
+30
View File
@@ -0,0 +1,30 @@
## Module Structure
```shell
.
├── README.md
├── cli
├── content
│ ├── pages
│ │ └── about
│ │ └── <static_page>.md
│ └── posts
│ └── <year>
│ └── <identifier>
│ ├── assets
│ │ └── <assets>
│ ├── translate_<lang>.md
│ └── index.md
├── requirements.txt
├── templates
│ ├── <template>.py
│ └── ...
└── utils
└── ...
```
### About `posts`
- Will render to `https://<DOMAIN>/blog/post/<identifier>`.
- Files under `assets/` (where images etc. located) should be pushed to OSS.
- For identifiers, use prefix `-` to indicate private (hidden) visibility. EX: `-hello-world`.
Executable
+27
View File
@@ -0,0 +1,27 @@
#!/usr/bin/env python3
import argparse
import rich
from templates.post import BlogPost
from utils.cache import init_cache
from utils.index import build_post_idx
def main():
# if args.new:
# new_post()
init_cache()
build_post_idx()
bp = BlogPost()
bp.init()
parser = argparse.ArgumentParser()
parser.add_argument("-n", "--new", help="Create a post with build-in template", type=str)
args = parser.parse_args(namespace=rich)
# todo: cli
if __name__ == '__main__':
main()
+4
View File
@@ -0,0 +1,4 @@
rich~=13.9.4
prompt_toolkit~=3.0.50
pypinyin~=0.53.0
python-frontmatter~=1.1.0
+87
View File
@@ -0,0 +1,87 @@
import uuid
from datetime import date
from datetime import datetime as dt
from typing import Dict, Optional
from templates.utils import Template, create_all_files
from utils.autocompletion import input_autocompletion, fuzzy_pinyin_matcher
from utils.cache import get_tags_cache, get_categories_cache
from utils.lang import eprint, wprint
POST_TEMPLATE = '''
---
title: {{ title }}
subtitle: {{ subtitle }}
published_date: {{ published_date }}
last_updated: {{ last_update }}
category: {{ category }}
tags: {{ tags }}
uuid: {{ uuid }}
---
WIP
'''
def get_title():
while True:
i = input("title > ")
if i.strip() == "":
eprint("empty title")
continue
else:
return i.strip()
def get_subtitle():
return input("subtitle > ")
def get_published_date():
while True:
d = input("published_date (YYYY-MM-DD) > ")
if d == "":
wprint("using today's date")
return dt.today().strftime("%Y-%m-%d")
else:
try:
_validate = dt.strptime(d, "%Y-%m-%d")
return d
except ValueError:
eprint("invalid date")
continue
def get_tags():
# todo: multi tags support
return input_autocompletion("tags > ", get_tags_cache(), fuzzy_pinyin_matcher)
def get_category():
while True:
i = input_autocompletion("category > ", get_categories_cache(), fuzzy_pinyin_matcher)
if i.strip() == "":
eprint("empty category")
continue
else:
return i.strip()
class BlogPost(Template):
WORK_PATH: str = "content/posts/{year}/{identifier}/"
ITEMS: Dict[str, Optional[str]] = {"index.md": POST_TEMPLATE, "assets/": None}
def init(self):
year = date.today().year
index_md = POST_TEMPLATE.format(
title=get_title(),
subtitle=get_subtitle(),
published_date=get_published_date(),
last_update=dt.today().strftime("%Y-%m-%d"),
tags=get_tags(),
category=get_category(),
uuid=uuid.uuid4()
)
create_all_files(self.WORK_PATH.format(year=year, identifier="test"), self.ITEMS)
+17
View File
@@ -0,0 +1,17 @@
import os
from abc import ABC, abstractmethod
from typing import Dict, Optional
class Template(ABC):
@abstractmethod
def init(self):
pass
def create_all_files(relative_work_path: str, items: Dict[str, Optional[str]]):
absolute_path = os.path.abspath(relative_work_path)
for item in items:
if item.endswith("/"):
print(absolute_path + item)
# todo: create files
+31
View File
@@ -0,0 +1,31 @@
class Ansi:
HEADER = '\033[95m'
BLUE = '\033[94m'
GREEN = '\033[92m'
WARNING = '\033[93m'
FAIL = '\033[91m'
BOLD = '\033[1m'
UNDERLINE = '\033[4m'
END = '\033[0m'
def disable(self) -> None:
self.HEADER = ''
self.BLUE = ''
self.GREEN = ''
self.WARNING = ''
self.FAIL = ''
self.BOLD = ''
self.UNDERLINE = ''
self.END = ''
@staticmethod
def blend(color: list[str], text: str) -> str:
return f"{"".join(color)}{text}{Ansi.END}"
@staticmethod
def bold(text: str) -> str:
return f"{Ansi.BOLD}{text}{Ansi.END}"
@staticmethod
def underline(text: str) -> str:
return f"{Ansi.UNDERLINE}{text}{Ansi.END}"
+46
View File
@@ -0,0 +1,46 @@
from collections.abc import Callable
from typing import List, Iterable
import prompt_toolkit
from prompt_toolkit.completion import Completer, CompleteEvent, Completion
from prompt_toolkit.document import Document
from pypinyin import lazy_pinyin, Style
class AutoCompleter(Completer):
def __init__(self, candidates: List[str], matching_algorithm: Callable[[str, str], bool]):
self.candidates = candidates
self.matching_algorithm = matching_algorithm
def get_completions(self, document: Document, complete_event: CompleteEvent) -> Iterable[Completion]:
curr_word = document.text_before_cursor
for c in self.candidates:
if self.matching_algorithm(curr_word, c):
yield Completion(c, start_position=-len(curr_word))
def exact_matcher(word: str, candidate: str, case_sensitive: bool = False) -> bool:
if case_sensitive:
return word == candidate
else:
return word.lower() == candidate.lower()
def fuzzy_matcher(word: str, candidate: str) -> bool:
return word.lower() in candidate.lower()
def fuzzy_pinyin_matcher(word: str, candidate: str) -> bool:
try:
word_matcher = ''.join(lazy_pinyin(word, style=Style.NORMAL))
candidate_matcher = ''.join(lazy_pinyin(candidate, style=Style.NORMAL))
return candidate_matcher.startswith(word_matcher)
except e:
return fuzzy_matcher(word, candidate)
def input_autocompletion(prompt: str, candidates: List[str],
matching_algorithm: Callable[[str, str], bool] = exact_matcher) -> str:
completer = AutoCompleter(candidates, matching_algorithm)
return prompt_toolkit.prompt(prompt, completer=completer, complete_in_thread=True)
+49
View File
@@ -0,0 +1,49 @@
import json
import os
from typing import TextIO
CACHE_DIR = ".cache"
ITEMS = {"register": "register.json"}
def overwrite_json_file(file: TextIO, json_data: dict):
file.seek(0)
file.write(json.dumps(json_data, indent=4, ensure_ascii=False))
file.truncate()
def init_cache():
if not os.path.exists(CACHE_DIR):
os.makedirs(CACHE_DIR)
for item in ITEMS.values():
item_path = os.path.join(CACHE_DIR, item)
if not os.path.exists(item_path):
with open(item_path, 'w') as cache_file:
match item:
case "register.json":
overwrite_json_file(cache_file, {})
break
def write_tags_cache(tags: list[str]):
with open(os.path.join(CACHE_DIR, ITEMS["register"]), 'r+') as tags_cache_file:
cache = json.load(tags_cache_file)
cache["tags"] = tags
overwrite_json_file(tags_cache_file, cache)
def write_category_cache(categories: list[str]):
with open(os.path.join(CACHE_DIR, ITEMS["register"]), 'r+') as categories_cache_file:
cache = json.load(categories_cache_file)
cache["categories"] = categories
overwrite_json_file(categories_cache_file, cache)
def get_tags_cache() -> list[str]:
with open(os.path.join(CACHE_DIR, ITEMS["register"]), 'r') as tags_cache_file:
return json.load(tags_cache_file)["tags"]
def get_categories_cache() -> list[str]:
with open(os.path.join(CACHE_DIR, ITEMS["register"]), 'r') as categories_cache_file:
return json.load(categories_cache_file)["categories"]
+27
View File
@@ -0,0 +1,27 @@
import os
import frontmatter
from utils.cache import write_tags_cache, write_category_cache
from utils.lang import list_exclude
EXCLUDE = [".DS_Store"]
POSTS_DIR = "content/posts"
def build_post_idx():
tags = set()
categories = set()
for year in list_exclude(os.listdir(POSTS_DIR), EXCLUDE):
for post_dir in list_exclude(os.listdir(os.path.join(POSTS_DIR, year)), EXCLUDE):
target_file = os.path.join(POSTS_DIR, year, post_dir, "index.md")
with open(target_file, "r") as f:
post = frontmatter.loads(f.read())
try:
for tag in post.metadata["tags"]:
tags.add(tag)
categories.add(post.metadata["category"])
except KeyError:
print(f"failed to parse {target_file}")
write_tags_cache(list(tags))
write_category_cache(list(categories))
+13
View File
@@ -0,0 +1,13 @@
from utils.ansi import Ansi
def list_exclude(source: list, exclude: list):
return filter(lambda item: item not in exclude, source)
def eprint(msg: str):
print(Ansi.blend([Ansi.FAIL], "error: ") + msg)
def wprint(msg: str):
print(Ansi.blend([Ansi.WARNING], "warn: ") + msg)
File renamed without changes.
@@ -1,5 +1,5 @@
[package]
name = "loadbalancer"
name = "gateway"
version = "0.1.0"
edition = "2021"
File renamed without changes.