4 Commits
Author SHA1 Message Date
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
36 changed files with 1827 additions and 38 deletions

No files matched your search

+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))
};
}
+1
View File
@@ -1 +1,2 @@
mod include_str;
mod log; mod log;
+16 -2
View File
@@ -5,10 +5,24 @@ edition = "2021"
[dependencies] [dependencies]
macros = { path = "../macros" } macros = { path = "../macros" }
markdown = { path = "../markdown" }
axum = { version = "0.8.1", features = ["macros"] }
tracing = "0.1.41" tracing = { version = "0.1.41", features = ["log"] }
git2 = "0.20.0" 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"] }
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"] }
dotenv = "0.15.0"
uuid = { version = "1.13.2", features = ["v4"] }
futures = "0.3.31"
[build-dependencies] [build-dependencies]
git2 = "0.20.0" git2 = "0.20.0"
+3 -2
View File
@@ -9,7 +9,8 @@ fn main() {
.peel_to_commit() .peel_to_commit()
.expect("failed to get HEAD commit"); .expect("failed to get HEAD commit");
let hash = commit.id().to_string(); let hash = commit.id().to_string();
let utc_date = Utc::now().date_naive().format("%Y%m%d").to_string(); let utc_timestamp = Utc::now().date_naive().format("%Y%m%d").to_string();
let build_id = format!("{}@{}", utc_date, &hash[..8]); 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); 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)
)
+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::{
filter::LevelFilter, fmt, fmt::format::Writer, fmt::time::FormatTime, fmt::MakeWriter,
layer::SubscriberExt, Layer, Registry,
};
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::{blog::Blog, ModuleTree};
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
}
+17 -1
View File
@@ -1,7 +1,23 @@
#[allow(dead_code)]
mod constants; mod constants;
#[allow(dead_code)]
mod scheduler;
mod init;
mod modules;
mod utils;
use crate::init::{logger, start};
const BUILD_ID: &str = env!("BUILD_ID"); const BUILD_ID: &str = env!("BUILD_ID");
fn main() { #[tokio::main]
async fn main() {
println!("{}", BUILD_ID); println!("{}", BUILD_ID);
init::load_env();
let _guards = logger::init();
start().await;
} }
@@ -0,0 +1,25 @@
use crate::constants::blog::PAGE_SIZE;
use axum::{extract::State, Json};
use serde_json::{json, Value};
use sqlx::{query, PgPool, Row};
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;
Whitespace-only changes.
Whitespace-only changes.
@@ -0,0 +1,30 @@
use axum::response::IntoResponse;
use axum::{
extract::{Path, State},
Json,
};
use serde_json::json;
use sqlx::PgPool;
const GET_POST_BY_I64_ID: &str = include_str!("get_post_by_i64_id.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 {
let post = if let Ok(id) = identifier.parse::<i64>() {
select_post_with_i64_id(&db_connection, id).await
} else {
select_post_with_str_id(&db_connection, &identifier).await
};
Json::from(json!({"res": "ok"})).into_response()
}
async fn select_post_with_i64_id(db: &PgPool, id: i64) {
todo!()
}
async fn select_post_with_str_id(db: &PgPool, id: &str) {
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,3 @@
pub mod get_page_count;
pub mod get_post;
// pub mod get_post_list;
@@ -0,0 +1,49 @@
use thiserror::Error;
#[derive(Error, Debug)]
pub enum PostIdxError {
#[error("missing essential field `{}` for file `{}`", field, filename)]
MissingField { field: String, filename: String },
#[error(
"invalid field `{}` for file `{}`, <{}>: {}",
field_name,
filename,
err_field,
msg
)]
InvalidField {
field_name: String,
err_field: String,
filename: String,
msg: String,
},
#[error(
"database error on writing `{}` to table `{}`: <{}>: {}",
data_desc,
db_table,
id,
msg
)]
DBWriteFailure {
data_desc: String,
db_table: String,
id: String,
msg: String,
},
#[error(
"Database error on Reading `{}` from table `{}`: <{}>: {}",
data_desc,
db_table,
id,
msg
)]
DBReadFailure {
data_desc: String,
db_table: String,
id: String,
msg: String,
},
}
@@ -0,0 +1,353 @@
use std::error::Error;
use std::path::Path;
use crate::{
modules::blog::cron::check_update::utils::get_md_file_basename,
modules::blog::cron::check_update::{error::PostIdxError, utils::parse_date},
};
use markdown::MdFile;
use sqlx::{query, PgPool, Pool, Postgres, Row};
use tokio::runtime::Runtime;
use tracing::{info, warn};
async fn base(db_connection: &PgPool, post: &MdFile) -> Result<(), PostIdxError> {
let post_id = post.file_id;
let title = match post.meta.get("title") {
None => {
return Err(PostIdxError::MissingField {
field: "title".to_string(),
filename: post.filename.clone(),
})
}
Some(title) => title.as_str().unwrap_or_else(|| {
warn!("invalid title for `{}`", &post.filename);
"invalid title"
}),
};
let sub_title = post
.meta
.get("sub_title")
.and_then(|v| v.as_str())
.unwrap_or("");
let summary = post
.meta
.get("summary")
.and_then(|v| v.as_str())
.unwrap_or("");
let content = post.content.as_str();
let category = match post.meta.get("category") {
None => {
return Err(PostIdxError::MissingField {
field: "category".to_string(),
filename: post.filename.clone(),
})
}
Some(title) => title.as_str().unwrap_or_else(|| {
warn!("invalid category for `{}`", &post.filename);
"invalid category"
}),
};
let last_update = {
if let Some(last_update) = post.meta.get("last_update") {
match parse_date(last_update.as_str().unwrap()) {
Ok(date) => Some(date),
Err(e) => {
return Err(PostIdxError::InvalidField {
field_name: "last_update".to_string(),
err_field: last_update.to_string(),
filename: post.filename.clone(),
msg: e.to_string(),
})
}
}
} else {
None
}
};
let first_update = match post.meta.get("date") {
None => {
return Err(PostIdxError::MissingField {
field: "date".to_string(),
filename: post.filename.clone(),
})
}
Some(first_update) => match parse_date(first_update.as_str().unwrap()) {
Ok(date) => date,
Err(e) => {
return Err(PostIdxError::InvalidField {
field_name: "date".to_string(),
err_field: first_update.to_string(),
filename: post.filename.clone(),
msg: e.to_string(),
})
}
},
};
let filename = get_md_file_basename(&post.filename);
if let Err(e) = query(
r##"
INSERT INTO Post (post_id, title, summary, content, last_update, first_update, sub_title, category, filename)
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,
filename = EXCLUDED.filename;
"##,
)
.bind(post_id)
.bind(title)
.bind(summary)
.bind(content)
.bind(last_update)
.bind(first_update)
.bind(sub_title)
.bind(category)
.bind(filename)
.execute(db_connection)
.await
{
return Err(
PostIdxError::DBWriteFailure {
data_desc: "post base information".to_string(),
db_table: "Post".to_string(),
id: post_id.to_string(),
msg: e.to_string(),
}
);
}
Ok(())
}
async fn tag(db_connection: &PgPool, post: &MdFile) -> Result<(), PostIdxError> {
if let Some(tags_string) = post.meta.get("tags").to_owned() {
let tags = match tags_string.as_array().map(|vec| {
vec.iter()
.map(|val| val.as_str().unwrap())
.collect::<Vec<_>>()
}) {
None => {
return Err(PostIdxError::MissingField {
field: "tags".to_string(),
filename: post.filename.clone(),
})
}
Some(val) => val,
};
let post_id = post.file_id;
// 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(),
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(),
msg: e.to_string(),
})
}
}
}
}
Ok(())
}
async fn hash(db_connection: &PgPool, post: &MdFile) -> Result<bool, PostIdxError> {
let base64_form = post.hash.clone();
match query("SELECT hash FROM Hash WHERE post_id = $1")
.bind(post.file_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.file_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.file_id.to_string(),
msg: e.to_string(),
});
}
Ok(true)
}
}
Err(e) => Err(PostIdxError::DBReadFailure {
data_desc: "post hash".to_string(),
db_table: "Hash".to_string(),
id: post.file_id.to_string(),
msg: e.to_string(),
}),
}
}
async fn meta(db_connection: &PgPool, post: &MdFile) -> Result<(), PostIdxError> {
if let Err(e) = query(
// todo: sanitization
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.file_id.to_string().as_str(), post.meta
).as_str()
)
.execute(db_connection)
.await {
Err(PostIdxError::DBWriteFailure {
data_desc: "post meta data".to_string(),
db_table: "Meta".to_string(),
id: post.file_id.to_string(),
msg: e.to_string(),
})
} else {
Ok(())
}
}
pub async fn index_file(path: &Path, db_conn: &Pool<Postgres>) -> Result<(), Box<dyn Error>> {
let md_file = MdFile::from_file(path)?;
async fn trail(db_conn: &Pool<Postgres>, md_file: &MdFile) -> Result<(), PostIdxError> {
base(db_conn, md_file).await?;
tag(db_conn, md_file).await?;
meta(db_conn, md_file).await?;
hash(db_conn, md_file).await.map(|_| ())
}
let mut cnt = 5;
while let Err(e) = trail(db_conn, &md_file).await {
match e {
PostIdxError::DBWriteFailure { .. } => {
cnt -= 1;
}
PostIdxError::DBReadFailure { .. } => {
cnt -= 1;
}
_ => return Err(e.into()),
}
if cnt <= 0 {
return Err(e.into());
}
}
info!("indexed markdown file: {}", path.display());
Ok(())
}
pub fn drop_index(file_basename: &str, db_conn: &Pool<Postgres>) -> Result<(), Box<dyn Error>> {
let rt = Runtime::new().unwrap();
rt.block_on(async {
if let Err(e) = query(
r#"
WITH deleted_post AS (
DELETE FROM post
WHERE id = (SELECT id FROM post WHERE filename = $1)
RETURNING id
)
DELETE FROM tag WHERE post_id IN (SELECT id FROM deleted_post);
DELETE FROM meta WHERE post_id IN (SELECT id FROM deleted_post);
DELETE FROM hash WHERE post_id IN (SELECT id FROM deleted_post);
"#,
)
.bind(file_basename)
.execute(db_conn)
.await
{
Err(Box::from(
format!("failed to drop index: {}", e).to_string(),
))
} else {
Ok(())
}
})
}
@@ -0,0 +1,148 @@
use std::{
env,
fs::remove_dir_all,
path::{Path, PathBuf},
};
use crate::{
constants::config::general::{default_value, var_name},
modules::blog::cron::check_update::{
index::drop_index,
utils::{expand_path, get_md_file_basename, index_files, search_md},
},
utils::git,
};
use macros::panic_with_log;
use git2::{Delta, Repository};
use sqlx::{Pool, Postgres};
use tracing::{error, info, warn, Level};
mod error;
mod index;
mod utils;
pub fn task(db_connection: &Pool<Postgres>) {
// 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());
match Repository::open(&git_work_dir) {
Ok(repo) => {
// prevent git remote url mismatch (clean cache)
match repo.find_remote(&git_remote_name) {
Ok(remote) => {
if remote.url().is_some_and(|url| url != git_url) {
remove_dir_all(&git_work_dir).unwrap();
warn!("blog content git({}) url mismatch, reinitialized", &git_url);
return;
}
}
Err(e) => {
remove_dir_all(&git_work_dir).unwrap();
error!(
"unexpected error on indexing blog content git({}), reinitialized: {}",
&git_url, e
);
}
}
match git::sync(&git_remote_name, &git_remote_branch, &repo) {
Ok(updates) => {
if !updates.is_empty() {
// incremental update
info!(
"blog content git({}) update detected: {:?}",
&git_url, updates
);
let mut update_md: Vec<PathBuf> = vec![];
for update in updates {
if [Delta::Deleted, Delta::Untracked].contains(&update.status) {
if let Some(path) = &update.old_path {
if drop_index(
&get_md_file_basename(
path.file_name().unwrap().to_str().unwrap(),
),
db_connection,
)
.is_ok()
{
info!(
"blog content git({}) file `{}` is detected as deleted/untracked, index dropped",
&git_url,
&path.display()
);
} else {
error!(
"blog content git({}) file `{}` is detected as deleted/untracked, failed to drop index",
&git_url,
&path.display()
)
}
}
} else if let Some(path) = &update.new_path {
if path.components().any(|comp| comp.as_os_str() == "posts")
&& path.extension().is_some_and(|ext| ext == "md")
{
update_md.push(path.to_path_buf());
}
}
}
index_files(&update_md, db_connection);
}
}
Err(e) => {
error!(
"failed to sync blog content git({}) updates: {}",
&git_url, e
);
}
}
}
Err(_) => {
match Repository::clone(&git_url, &git_work_dir) {
Ok(_repo) => {
// full repo index
let files = search_md(git_work_dir.join("posts"));
index_files(&files, db_connection);
}
Err(e) => {
error!("failed to clone blog content git({}): {}", &git_url, e);
}
}
}
};
}
@@ -0,0 +1,88 @@
use std::path::PathBuf;
use crate::{
constants::{
config::general::{default_value, var_name},
time::DEFAULT_DATE_FORMAT,
},
modules::blog::cron::check_update::index::index_file,
};
use chrono::{DateTime, FixedOffset, ParseResult};
use dirs::home_dir;
use futures::{executor::block_on, future::join_all};
use sqlx::{Pool, Postgres};
use tokio::{runtime::Runtime, spawn};
use tracing::{error, 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()
.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_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", "")
}
@@ -0,0 +1 @@
// pub mod check_update;
+52
View File
@@ -0,0 +1,52 @@
mod api;
mod cron;
use crate::{
modules::{
blog::{
api::{
get_page_count,
get_post,
// get_post_list::get_post_list,
},
// cron::check_update,
},
Module,
},
scheduler::tasks::CronTask,
};
use axum::{routing::get, Router};
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<Box<dyn CronTask>> {
// let db_handler = self.db_handler.clone();
vec![
// BasicTask::new(move || check_update::task(&db_handler), "0/15 * * * * *")
// .unwrap()
// .to_task(),
]
}
}
@@ -0,0 +1,41 @@
CREATE TABLE tags
(
id SERIAL PRIMARY KEY,
name VARCHAR(20) UNIQUE NOT NULL,
description TEXT
);
CREATE TABLE categories
(
id SERIAL PRIMARY KEY,
name VARCHAR(20) UNIQUE NOT NULL,
description TEXT
);
CREATE TABLE posts
(
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 categories (id) NOT NULL
);
CREATE TABLE post_tag
(
post SERIAL REFERENCES posts (id),
tag SERIAL REFERENCES tags (id)
);
CREATE TABLE metadata
(
pid SERIAL REFERENCES posts (id) NOT NULL,
meta_key VARCHAR(50) NOT NULL,
meta_value TEXT NOT NULL,
CONSTRAINT meta_pk PRIMARY KEY (pid, meta_key)
);
+84
View File
@@ -0,0 +1,84 @@
pub mod blog;
use std::env;
use std::num::ParseIntError;
use crate::{
constants::config::server::{default_value, var_name},
scheduler::{tasks::CronTask, Scheduler},
};
use macros::error_panic;
use axum::Router;
use tracing::{info, warn};
// 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<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 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 tokio::net::TcpListener::bind(format!(
"{}:{}",
default_value::HOST,
port
))
.await
{
Ok(listener) => {
info!("server started 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());
}
axum::serve(listener, app).await.unwrap_or_else(|e| {
error_panic!("failed to start axum: {}", e.to_string());
});
}
}
+202
View File
@@ -0,0 +1,202 @@
use std::{
collections::HashMap,
sync::{
atomic::{AtomicBool, Ordering},
Arc, RwLock,
},
thread,
};
use crate::scheduler::{
scheduler_error::{
SchedulerError, SchedulerError::IdAlreadyRunning, SchedulerError::IdNotFound,
},
sleeper::SkippableSleeper,
tasks::CronTask,
};
use tracing::log::error;
use uuid::Uuid;
pub mod scheduler_error;
mod sleeper;
mod task_func;
pub mod tasks;
struct TaskStatus {
sleeper: Option<Arc<SkippableSleeper>>,
is_alive: AtomicBool,
}
#[derive(Default)]
pub struct Scheduler {
tasks: Arc<RwLock<HashMap<Uuid, Arc<Box<dyn CronTask>>>>>,
running: Arc<RwLock<HashMap<Uuid, Arc<TaskStatus>>>>,
}
impl Scheduler {
/// Start all tasks in the scheduler
pub fn start(&self) -> Result<(), SchedulerError> {
for (k, _v) in self.tasks.read()?.iter() {
self.run_non_blocking(k)?
}
Ok(())
}
/// Terminate all tasks in the scheduler
pub fn halt() {}
fn run_task(
running: Arc<RwLock<HashMap<Uuid, Arc<TaskStatus>>>>,
id: Uuid,
task: Arc<Box<dyn CronTask>>,
) -> Result<(), SchedulerError> {
if running.read()?.get(&id).is_some() {
return Err(IdAlreadyRunning);
} else {
running.write()?.insert(
id,
TaskStatus {
sleeper: None,
is_alive: AtomicBool::new(true),
}
.into(),
);
}
loop {
if let Some(next_execution) = task.get_next_execution() {
// guarantee is alive
let task_status = {
let running_reading_guard = running.read()?;
running_reading_guard.get(&id).cloned().clone()
};
if let Some(task_status) = task_status {
if !&task_status.is_alive.load(Ordering::SeqCst) {
{
let mut running_lock_guard = running.write()?;
running_lock_guard.remove(&id);
}
return Ok(());
} else {
let sleeper = Arc::from(SkippableSleeper::from(next_execution));
{
let mut running_lock_guard = running.write()?;
running_lock_guard.insert(
id,
TaskStatus {
sleeper: Some(sleeper.clone()),
is_alive: AtomicBool::new(true),
}
.into(),
);
}
sleeper.start();
task.call();
}
}
} else {
error!("unexpected error ");
break;
}
}
Ok(())
}
/// Run a task with `id`
pub fn run(&self, id: &Uuid) -> Result<(), SchedulerError> {
let id = id.to_owned();
if let Some(task) = self.tasks.read()?.get(&id) {
Self::run_task(self.running.clone(), id, task.clone())
} else {
Err(IdNotFound)
}
}
pub fn run_non_blocking(&self, id: &Uuid) -> Result<(), SchedulerError> {
let id = id.to_owned();
if let Some(task) = self.tasks.read()?.get(&id) {
let running_clone = self.running.clone();
let task_clone = task.clone();
thread::spawn(move || Self::run_task(running_clone, id, task_clone));
Ok(())
} else {
Err(IdNotFound)
}
}
fn get_running(&self, id: &Uuid) -> Option<Arc<TaskStatus>> {
let ret = self.running.read().unwrap().get(id).cloned();
ret
}
/// Stop a task with `id`
pub fn stop(&self, id: &Uuid) -> Result<(), SchedulerError> {
match self.get_running(id) {
None => Err(IdNotFound),
Some(status) => {
status.is_alive.store(false, Ordering::SeqCst);
if let Some(s) = &status.sleeper {
s.skip();
} else {
return Err(IdAlreadyRunning);
}
Ok(())
}
}
}
pub fn now(&self, id: &Uuid) -> Result<(), SchedulerError> {
match self.get_running(id) {
None => Err(IdNotFound),
Some(status) => {
if let Some(s) = &status.sleeper {
s.skip()
} else {
return Err(IdAlreadyRunning);
}
Ok(())
}
}
}
pub fn insert(&self, task: Box<dyn CronTask>) -> Result<Uuid, SchedulerError> {
let id = Uuid::new_v4();
self.tasks.write()?.insert(id, Arc::new(task));
Ok(id)
}
pub fn insert_list(&self, tasks: Vec<Box<dyn CronTask>>) -> Result<Vec<Uuid>, SchedulerError> {
tasks.into_iter().map(|t| self.insert(t)).collect()
}
pub fn new() -> Self {
Self::default()
}
}
#[cfg(test)]
mod tests {
use crate::scheduler::tasks::basic::BasicTask;
use crate::scheduler::Scheduler;
use std::error::Error;
use std::thread;
use std::time::Duration;
#[test]
fn test_run() -> Result<(), Box<dyn Error>> {
let scheduler = Scheduler::default();
let task = BasicTask::new(|| println!("test print"), "* * * * * *")?;
let id = scheduler.insert(Box::new(task))?;
for i in 1..=2 {
println!("Loop {} start", i);
println!("Run {}", i);
scheduler.run_non_blocking(&id)?;
thread::sleep(Duration::from_secs(4));
scheduler.stop(&id)?;
println!("Stop {}", i);
thread::sleep(Duration::from_secs(2));
println!("Loop {} finished", i);
}
Ok(())
}
}
@@ -0,0 +1,48 @@
use std::{
error::Error,
fmt::{Debug, Display, Formatter},
sync::PoisonError,
};
use crate::scheduler::scheduler_error::SchedulerError::MutexPoison;
// todo: refactor with thiserror
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 {}
+80
View File
@@ -0,0 +1,80 @@
use std::{
sync::{Arc, Condvar, Mutex},
thread,
time::Duration,
};
use chrono::{DateTime, Utc};
#[derive(Clone, Debug, Default)]
pub struct SkippableSleeper {
pub duration: Duration,
state: Arc<(Mutex<bool>, Condvar)>,
}
impl SkippableSleeper {
pub fn new(duration: Duration) -> Self {
SkippableSleeper {
duration,
state: Arc::new((Mutex::new(false), Condvar::new())),
}
}
// Blocks until the duration has passed or skip() is called
pub fn start(&self) {
let (lock, cvar) = &*self.state;
let mut received = lock.lock().unwrap();
let state_clone = Arc::clone(&self.state);
let duration_clone = self.duration;
thread::spawn(move || {
thread::sleep(duration_clone);
let (lock, cvar) = &*state_clone;
let mut received = lock.lock().unwrap();
*received = true;
cvar.notify_one();
});
while !*received {
received = cvar.wait(received).unwrap();
}
}
// Skip the blocking delay
pub fn skip(&self) {
let (lock, cvar) = &*self.state;
let mut received = lock.lock().unwrap();
*received = true;
cvar.notify_one();
}
}
impl From<DateTime<Utc>> for SkippableSleeper {
fn from(value: DateTime<Utc>) -> Self {
let from_utc_timestamp = value.timestamp() as u64;
let curr_utc_timestamp = Utc::now().timestamp() as u64;
SkippableSleeper::new(Duration::from_secs(from_utc_timestamp - curr_utc_timestamp))
}
}
#[cfg(test)]
mod test {
use crate::scheduler::sleeper::SkippableSleeper;
use std::thread;
use std::time::Duration;
#[test]
fn usage() {
let sleeper = SkippableSleeper::new(Duration::new(10, 0));
thread::spawn({
let sleeper = sleeper.clone();
move || {
println!("Sleeping...");
sleeper.start();
println!("Woke up!");
}
});
println!("Start Sleeping...");
thread::sleep(Duration::from_secs(4));
println!("Skipping sleep...");
sleeper.skip();
}
}
@@ -0,0 +1,3 @@
pub trait TaskFunc: Fn() + Send + Sync + 'static {}
impl<T> TaskFunc for T where T: Fn() + Send + Sync + Clone + 'static {}
@@ -0,0 +1,38 @@
use std::error::Error;
use std::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) {
(self.func)()
}
}
@@ -0,0 +1,8 @@
use chrono::{DateTime, Utc};
pub mod basic;
pub trait CronTask: Send + Sync {
fn get_next_execution(&self) -> Option<DateTime<Utc>>;
fn call(&self);
}
+166
View File
@@ -0,0 +1,166 @@
/// git merge
/// 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::{
build, AnnotatedCommit, AutotagOption, Delta, FetchOptions, Reference, Remote, Repository,
};
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,
}
#[allow(dead_code)]
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)
}
#[allow(dead_code)]
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());
// println!("{}", msg);
lb.set_target(rc.id(), &msg)?;
repo.set_head(&name)?;
repo.checkout_head(Some(build::CheckoutBuilder::default().force()))?;
Ok(())
}
#[allow(dead_code)]
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(())
}
#[allow(dead_code)]
fn do_merge<'a>(
repo: &'a Repository,
remote_branch: &str,
fetch_commit: AnnotatedCommit<'a>,
) -> Result<Vec<FileDelta>, 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)?;
}
calculate_diff(repo, &head_commit, &fetch_commit)
}
#[allow(dead_code)]
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)?;
do_merge(repo, remote_branch, fetch_commit)
}
+1
View File
@@ -0,0 +1 @@
pub mod git;
File renamed without changes.
@@ -1,5 +1,5 @@
[package] [package]
name = "loadbalancer" name = "gateway"
version = "0.1.0" version = "0.1.0"
edition = "2021" edition = "2021"
File renamed without changes.