feat: file history endpoint
- chore: cleaning some unused Signed-off-by: Pakin <pakin.t@forth.co.th>
This commit is contained in:
Pakin
parent
2466472235
commit
4709c3b2b7
10 changed files with 313 additions and 1244 deletions
238
Cargo.lock
generated
238
Cargo.lock
generated
|
|
@ -506,9 +506,9 @@ dependencies = [
|
|||
|
||||
[[package]]
|
||||
name = "chrono"
|
||||
version = "0.4.44"
|
||||
version = "0.4.45"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "c673075a2e0e5f4a1dde27ce9dee1ea4558c7ffe648f576438a20ca1d2acc4b0"
|
||||
checksum = "1aa79e62e7697b8e29b513a68abacf485adcd1fe8284a4316c5ae868e6633327"
|
||||
dependencies = [
|
||||
"iana-time-zone",
|
||||
"js-sys",
|
||||
|
|
@ -937,12 +937,6 @@ version = "0.1.9"
|
|||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "5baebc0774151f905a1a2cc41989300b1e6fbb29aff0ceffa1064fdd3088d582"
|
||||
|
||||
[[package]]
|
||||
name = "fixedbitset"
|
||||
version = "0.5.7"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "1d674e81391d1e1ab681a28d99df07927c6d4aa5b027d7da16ba32d1d21ecd99"
|
||||
|
||||
[[package]]
|
||||
name = "flate2"
|
||||
version = "1.1.9"
|
||||
|
|
@ -1275,19 +1269,6 @@ dependencies = [
|
|||
"tower-service",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "hyper-timeout"
|
||||
version = "0.5.2"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "2b90d566bffbce6a75bd8b09a05aa8c2cb1fabb6cb348f8840c9e4c90a0d83b0"
|
||||
dependencies = [
|
||||
"hyper",
|
||||
"hyper-util",
|
||||
"pin-project-lite",
|
||||
"tokio",
|
||||
"tower-service",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "hyper-tls"
|
||||
version = "0.6.0"
|
||||
|
|
@ -1867,12 +1848,6 @@ dependencies = [
|
|||
"version_check",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "multimap"
|
||||
version = "0.10.1"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "1d87ecb2933e8aeadb3e3a02b828fed80a7528047e68b4f424523a0981a3a084"
|
||||
|
||||
[[package]]
|
||||
name = "native-tls"
|
||||
version = "0.2.18"
|
||||
|
|
@ -2112,17 +2087,6 @@ version = "2.3.2"
|
|||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "9b4f627cb1b25917193a259e49bdad08f671f8d9708acfd5fe0a8c1455d87220"
|
||||
|
||||
[[package]]
|
||||
name = "petgraph"
|
||||
version = "0.8.3"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "8701b58ea97060d5e5b155d383a69952a60943f0e6dfe30b04c287beb0b27455"
|
||||
dependencies = [
|
||||
"fixedbitset",
|
||||
"hashbrown 0.15.5",
|
||||
"indexmap",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "phf"
|
||||
version = "0.13.1"
|
||||
|
|
@ -2142,26 +2106,6 @@ dependencies = [
|
|||
"siphasher",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "pin-project"
|
||||
version = "1.1.12"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "cbf0d9e68100b3a7989b4901972f265cd542e560a3a8a724e1e20322f4d06ce9"
|
||||
dependencies = [
|
||||
"pin-project-internal",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "pin-project-internal"
|
||||
version = "1.1.12"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "a990e22f43e84855daf260dded30524ef4a9021cc7541c26540500a50b624389"
|
||||
dependencies = [
|
||||
"proc-macro2",
|
||||
"quote",
|
||||
"syn",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "pin-project-lite"
|
||||
version = "0.2.17"
|
||||
|
|
@ -2305,79 +2249,6 @@ dependencies = [
|
|||
"syn",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "prost"
|
||||
version = "0.14.3"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "d2ea70524a2f82d518bce41317d0fae74151505651af45faf1ffbd6fd33f0568"
|
||||
dependencies = [
|
||||
"bytes",
|
||||
"prost-derive",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "prost-build"
|
||||
version = "0.14.3"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "343d3bd7056eda839b03204e68deff7d1b13aba7af2b2fd16890697274262ee7"
|
||||
dependencies = [
|
||||
"heck",
|
||||
"itertools",
|
||||
"log",
|
||||
"multimap",
|
||||
"petgraph",
|
||||
"prettyplease",
|
||||
"prost",
|
||||
"prost-types",
|
||||
"pulldown-cmark",
|
||||
"pulldown-cmark-to-cmark",
|
||||
"regex",
|
||||
"syn",
|
||||
"tempfile",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "prost-derive"
|
||||
version = "0.14.3"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "27c6023962132f4b30eb4c172c91ce92d933da334c59c23cddee82358ddafb0b"
|
||||
dependencies = [
|
||||
"anyhow",
|
||||
"itertools",
|
||||
"proc-macro2",
|
||||
"quote",
|
||||
"syn",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "prost-types"
|
||||
version = "0.14.3"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "8991c4cbdb8bc5b11f0b074ffe286c30e523de90fee5ba8132f1399f23cb3dd7"
|
||||
dependencies = [
|
||||
"prost",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "pulldown-cmark"
|
||||
version = "0.13.3"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "7c3a14896dfa883796f1cb410461aef38810ea05f2b2c33c5aded3649095fdad"
|
||||
dependencies = [
|
||||
"bitflags",
|
||||
"memchr",
|
||||
"unicase",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "pulldown-cmark-to-cmark"
|
||||
version = "22.0.0"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "50793def1b900256624a709439404384204a5dc3a6ec580281bfaac35e882e90"
|
||||
dependencies = [
|
||||
"pulldown-cmark",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "pxfm"
|
||||
version = "0.1.29"
|
||||
|
|
@ -3057,6 +2928,7 @@ dependencies = [
|
|||
"bb8-redis",
|
||||
"brotli",
|
||||
"bytes",
|
||||
"chrono",
|
||||
"env_logger",
|
||||
"futures-util",
|
||||
"git2",
|
||||
|
|
@ -3065,7 +2937,6 @@ dependencies = [
|
|||
"libgit2-sys",
|
||||
"libtbr",
|
||||
"log",
|
||||
"prost",
|
||||
"redis",
|
||||
"reqwest",
|
||||
"serde",
|
||||
|
|
@ -3073,9 +2944,6 @@ dependencies = [
|
|||
"tokio",
|
||||
"tokio-postgres",
|
||||
"tokio-util",
|
||||
"tonic",
|
||||
"tonic-prost",
|
||||
"tonic-prost-build",
|
||||
"uuid",
|
||||
]
|
||||
|
||||
|
|
@ -3244,17 +3112,6 @@ dependencies = [
|
|||
"tokio",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "tokio-stream"
|
||||
version = "0.1.18"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "32da49809aab5c3bc678af03902d4ccddea2a87d028d86392a4b1560c6906c70"
|
||||
dependencies = [
|
||||
"futures-core",
|
||||
"pin-project-lite",
|
||||
"tokio",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "tokio-util"
|
||||
version = "0.7.18"
|
||||
|
|
@ -3268,74 +3125,6 @@ dependencies = [
|
|||
"tokio",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "tonic"
|
||||
version = "0.14.6"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "ac2a5518c70fa84342385732db33fb3f44bc4cc748936eb5833d2df34d6445ef"
|
||||
dependencies = [
|
||||
"async-trait",
|
||||
"axum",
|
||||
"base64",
|
||||
"bytes",
|
||||
"h2",
|
||||
"http",
|
||||
"http-body",
|
||||
"http-body-util",
|
||||
"hyper",
|
||||
"hyper-timeout",
|
||||
"hyper-util",
|
||||
"percent-encoding",
|
||||
"pin-project",
|
||||
"socket2",
|
||||
"sync_wrapper",
|
||||
"tokio",
|
||||
"tokio-stream",
|
||||
"tower",
|
||||
"tower-layer",
|
||||
"tower-service",
|
||||
"tracing",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "tonic-build"
|
||||
version = "0.14.6"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "c68f61875ac5293cf72e6c8cf0158086428c82c37229e98c840878f1706b0322"
|
||||
dependencies = [
|
||||
"prettyplease",
|
||||
"proc-macro2",
|
||||
"quote",
|
||||
"syn",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "tonic-prost"
|
||||
version = "0.14.6"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "50849f68853be452acf590cde0b146665b8d507b3b8af17261df47e02c209ea0"
|
||||
dependencies = [
|
||||
"bytes",
|
||||
"prost",
|
||||
"tonic",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "tonic-prost-build"
|
||||
version = "0.14.6"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "654e5643eff75d7f8c99197ce1440ed19a3474eada74c12bbac488b2cafdae27"
|
||||
dependencies = [
|
||||
"prettyplease",
|
||||
"proc-macro2",
|
||||
"prost-build",
|
||||
"prost-types",
|
||||
"quote",
|
||||
"syn",
|
||||
"tempfile",
|
||||
"tonic-build",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "tower"
|
||||
version = "0.5.3"
|
||||
|
|
@ -3344,12 +3133,9 @@ checksum = "ebe5ef63511595f1344e2d5cfa636d973292adc0eec1f0ad45fae9f0851ab1d4"
|
|||
dependencies = [
|
||||
"futures-core",
|
||||
"futures-util",
|
||||
"indexmap",
|
||||
"pin-project-lite",
|
||||
"slab",
|
||||
"sync_wrapper",
|
||||
"tokio",
|
||||
"tokio-util",
|
||||
"tower-layer",
|
||||
"tower-service",
|
||||
"tracing",
|
||||
|
|
@ -3393,21 +3179,9 @@ checksum = "63e71662fa4b2a2c3a26f570f037eb95bb1f85397f3cd8076caed2f026a6d100"
|
|||
dependencies = [
|
||||
"log",
|
||||
"pin-project-lite",
|
||||
"tracing-attributes",
|
||||
"tracing-core",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "tracing-attributes"
|
||||
version = "0.1.31"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "7490cfa5ec963746568740651ac6781f701c9c5ea257c58e057f3ba8cf69e8da"
|
||||
dependencies = [
|
||||
"proc-macro2",
|
||||
"quote",
|
||||
"syn",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "tracing-core"
|
||||
version = "0.1.36"
|
||||
|
|
@ -3429,12 +3203,6 @@ version = "1.20.0"
|
|||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "40ce102ab67701b8526c123c1bab5cbe42d7040ccfd0f64af1a385808d2f43de"
|
||||
|
||||
[[package]]
|
||||
name = "unicase"
|
||||
version = "2.9.0"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "dbc4bc3a9f746d862c45cb89d705aa10f187bb96c76001afab07a0d35ce60142"
|
||||
|
||||
[[package]]
|
||||
name = "unicode-bidi"
|
||||
version = "0.3.18"
|
||||
|
|
|
|||
|
|
@ -20,17 +20,12 @@ json-patch = "4.1.0"
|
|||
libgit2-sys = { version = "0.18.3", features = ["ssh"] }
|
||||
libtbr = { git = "https://pakin-inspiron-15-3530.tail110d9.ts.net/pakin/libtbr.git", version = "0.1.1" }
|
||||
log = "0.4.29"
|
||||
prost = "0.14.1"
|
||||
redis = { version = "1.0.2", features = ["tokio-comp"] }
|
||||
reqwest = { version = "0.12.25", features = ["json"] }
|
||||
serde = { version = "1.0.228", features = ["derive"] }
|
||||
serde_json = { version = "1.0.145", features = ["preserve_order"] }
|
||||
tokio = { version = "1.48.0", features = ["full"] }
|
||||
tokio-util = { version = "0.7.18", features = ["io"] }
|
||||
tonic = { version = "0.14.2", features = ["transport"] }
|
||||
tonic-prost = "0.14.2"
|
||||
uuid = { version = "1.20.0", features = ["v4"] }
|
||||
tokio-postgres = "0.7.17"
|
||||
|
||||
[build-dependencies]
|
||||
tonic-prost-build = "0.14.2"
|
||||
chrono = "0.4.45"
|
||||
|
|
|
|||
5
build.rs
5
build.rs
|
|
@ -1,5 +0,0 @@
|
|||
fn main() -> Result<(), Box<dyn std::error::Error>> {
|
||||
tonic_prost_build::configure().compile_protos(&["tbm-proto/registry.proto"], &["tbm-proto"])?;
|
||||
|
||||
Ok(())
|
||||
}
|
||||
509
src/app.rs
509
src/app.rs
|
|
@ -1,12 +1,9 @@
|
|||
use std::{
|
||||
collections::HashMap,
|
||||
io::Cursor,
|
||||
path::{Path, PathBuf},
|
||||
sync::Arc,
|
||||
time::Duration,
|
||||
};
|
||||
|
||||
use async_compression::tokio::bufread::BrotliEncoder;
|
||||
use axum::{
|
||||
Json, Router,
|
||||
body::Body,
|
||||
|
|
@ -15,51 +12,30 @@ use axum::{
|
|||
routing::{get, post},
|
||||
};
|
||||
use axum_extra::extract::multipart::Multipart;
|
||||
use axum_macros::debug_handler;
|
||||
use bb8::{Pool, PooledConnection};
|
||||
use bb8::Pool;
|
||||
use bb8_redis::RedisConnectionManager;
|
||||
use brotli::CompressorWriter;
|
||||
use git2::{
|
||||
Cred, FetchOptions, Object, ObjectType, Oid, PushOptions, RemoteCallbacks, Repository,
|
||||
ResetType, build::CheckoutBuilder,
|
||||
Cred, FetchOptions, ObjectType, Oid, RemoteCallbacks, Repository, ResetType,
|
||||
build::CheckoutBuilder,
|
||||
};
|
||||
|
||||
use image::load_from_memory;
|
||||
use log::{error, info, warn};
|
||||
use redis::{AsyncTypedCommands, Connection, TypedCommands, aio::MultiplexedConnection};
|
||||
use reqwest::header;
|
||||
use serde::{Deserialize, Serialize};
|
||||
use serde_json::{Value, json};
|
||||
use tokio::{
|
||||
io::{AsyncReadExt, BufReader},
|
||||
sync::{
|
||||
Mutex, MutexGuard,
|
||||
mpsc::{Receiver, Sender, channel},
|
||||
},
|
||||
task::JoinSet,
|
||||
};
|
||||
use tokio::sync::{Mutex, MutexGuard};
|
||||
|
||||
use tokio_postgres::NoTls;
|
||||
use uuid::Uuid;
|
||||
|
||||
use crate::{
|
||||
gcm,
|
||||
tx::{
|
||||
self,
|
||||
helpers::{doc_snap_key, doc_tx_stream_key, doc_ver_key, now_unix},
|
||||
patcher::apply_ops,
|
||||
types::PatchOp,
|
||||
},
|
||||
};
|
||||
use crate::gcm;
|
||||
|
||||
type RedisPool = Pool<RedisConnectionManager>;
|
||||
|
||||
#[derive(Debug, Clone)]
|
||||
pub struct CacheJob {
|
||||
pub rel_path: String,
|
||||
pub file_data: Option<Vec<u8>>,
|
||||
}
|
||||
|
||||
// #[derive(Debug, Clone)]
|
||||
// pub struct CacheJob {
|
||||
// pub rel_path: String,
|
||||
// pub file_data: Option<Vec<u8>>,
|
||||
// }
|
||||
#[derive(Clone)]
|
||||
pub struct AppState {
|
||||
// cached_country_names: Vec<&'static str>,
|
||||
|
|
@ -67,9 +43,9 @@ pub struct AppState {
|
|||
repo: Arc<Mutex<Repository>>,
|
||||
// commit external db
|
||||
postgres: Arc<Mutex<tokio_postgres::Client>>,
|
||||
redis: RedisPool,
|
||||
_redis: RedisPool,
|
||||
// save already fetched path, further calls should fetch from redis instead.
|
||||
fetched: Vec<String>,
|
||||
_fetched: Vec<String>,
|
||||
}
|
||||
|
||||
impl AppState {
|
||||
|
|
@ -88,12 +64,6 @@ impl AppState {
|
|||
|
||||
const SUPPORTED_IMAGES: [&str; 6] = ["jpg", "jpeg", "png", "gif", "bmp", "webp"];
|
||||
|
||||
fn is_supported_image_file_type(name: String) -> bool {
|
||||
let ext = name.split(".").last().unwrap_or("");
|
||||
|
||||
SUPPORTED_IMAGES.contains(&ext)
|
||||
}
|
||||
|
||||
fn detect_content_type(path: &str) -> String {
|
||||
let ext = path.split(".").last().unwrap_or("");
|
||||
if ext.eq(".json") {
|
||||
|
|
@ -123,44 +93,8 @@ async fn checkout_handler(
|
|||
param.path
|
||||
);
|
||||
|
||||
// println!(
|
||||
// "req: {param:?}, has_cached: {}, saved: {:?}",
|
||||
// state
|
||||
// .fetched
|
||||
// .contains(¶m.path.clone().as_str().to_string()),
|
||||
// state.fetched.clone()
|
||||
// );
|
||||
|
||||
// quick response from redis
|
||||
|
||||
let repo = state.repo.clone();
|
||||
|
||||
// let obj = match repo.revparse_single(&fpath) {
|
||||
// Ok(obj) => obj,
|
||||
// Err(e) => {
|
||||
// let error_log = format!("unexpected revparse single: {err}", err = e.message());
|
||||
// error!("{error_log}");
|
||||
// response.insert("error".to_string(), json!(error_log.clone()));
|
||||
// return (
|
||||
// axum::http::StatusCode::INTERNAL_SERVER_ERROR,
|
||||
// Json(json!(response)),
|
||||
// )
|
||||
// .into_response();
|
||||
// }
|
||||
// };
|
||||
//
|
||||
|
||||
// let obj_bytes = match get_git_object(repo, &fpath).await {
|
||||
// Ok(ob) => ob,
|
||||
// Err(e) => {
|
||||
// return (
|
||||
// axum::http::StatusCode::INTERNAL_SERVER_ERROR,
|
||||
// Json(json!(format!("{e}"))),
|
||||
// )
|
||||
// .into_response();
|
||||
// }
|
||||
// };
|
||||
|
||||
let fpath_clone = fpath.clone();
|
||||
let result = tokio::task::spawn_blocking(move || {
|
||||
let repo = repo.try_lock()?;
|
||||
|
|
@ -171,10 +105,6 @@ async fn checkout_handler(
|
|||
|
||||
match result {
|
||||
Ok(Ok(o)) => {
|
||||
let state_clone = state.clone();
|
||||
let key_clone = fpath.clone();
|
||||
let data_clone = o.clone();
|
||||
|
||||
let content_type = detect_content_type(&fpath);
|
||||
Response::builder()
|
||||
.header(header::CONTENT_TYPE, content_type)
|
||||
|
|
@ -250,7 +180,7 @@ async fn checkout_handler(
|
|||
|
||||
async fn fetch_handler(State(state): State<AppState>) -> impl IntoResponse {
|
||||
let mut response: HashMap<String, Value> = HashMap::new();
|
||||
if let Some(repo_path) = state.clone().get_config("GIT_REPO_LOCAL_DEST") {
|
||||
if let Some(_repo_path) = state.clone().get_config("GIT_REPO_LOCAL_DEST") {
|
||||
let repo = state.repo.lock().await;
|
||||
|
||||
let mut remote = match repo.find_remote("origin") {
|
||||
|
|
@ -336,19 +266,19 @@ async fn fetch_handler(State(state): State<AppState>) -> impl IntoResponse {
|
|||
// { path: "/path/to/file", signature: {
|
||||
// username: "", email: ""
|
||||
// }, patch_key: "edit_id_from_redis"}
|
||||
#[derive(Deserialize)]
|
||||
struct CommitBody {
|
||||
// Actual file path
|
||||
path: String,
|
||||
// Signature of user
|
||||
signature: Signature,
|
||||
// Key to grep content changes from redis
|
||||
patch_key: String,
|
||||
// user message
|
||||
message: Option<String>,
|
||||
#[serde(flatten)]
|
||||
extra: HashMap<String, Value>,
|
||||
}
|
||||
// #[derive(Deserialize)]
|
||||
// struct CommitBody {
|
||||
// // Actual file path
|
||||
// path: String,
|
||||
// // Signature of user
|
||||
// signature: Signature,
|
||||
// // Key to grep content changes from redis
|
||||
// patch_key: String,
|
||||
// // user message
|
||||
// message: Option<String>,
|
||||
// #[serde(flatten)]
|
||||
// extra: HashMap<String, Value>,
|
||||
// }
|
||||
|
||||
#[derive(Deserialize, Debug)]
|
||||
struct Signature {
|
||||
|
|
@ -628,34 +558,34 @@ async fn commit_handler(
|
|||
)
|
||||
}
|
||||
|
||||
async fn fetch_content_from_redis(redis: RedisPool, key: &str) -> Result<String, String> {
|
||||
match redis.get().await {
|
||||
Ok(mut s) => {
|
||||
if let Ok(res) = s.get(key).await
|
||||
&& let Some(res) = res
|
||||
{
|
||||
Ok(res)
|
||||
} else {
|
||||
Err(format!("result error from key: {key}"))
|
||||
}
|
||||
}
|
||||
Err(e) => Err(format!("redis get failed: {e}")),
|
||||
}
|
||||
}
|
||||
// async fn fetch_content_from_redis(redis: RedisPool, key: &str) -> Result<String, String> {
|
||||
// match redis.get().await {
|
||||
// Ok(mut s) => {
|
||||
// if let Ok(res) = s.get(key).await
|
||||
// && let Some(res) = res
|
||||
// {
|
||||
// Ok(res)
|
||||
// } else {
|
||||
// Err(format!("result error from key: {key}"))
|
||||
// }
|
||||
// }
|
||||
// Err(e) => Err(format!("redis get failed: {e}")),
|
||||
// }
|
||||
// }
|
||||
|
||||
async fn fetch_content_from_redis_byte(
|
||||
mut conn: MultiplexedConnection,
|
||||
key: &str,
|
||||
) -> Result<Vec<u8>, String> {
|
||||
let res = redis::cmd("GET").arg(key).query_async(&mut conn).await;
|
||||
// async fn fetch_content_from_redis_byte(
|
||||
// mut conn: MultiplexedConnection,
|
||||
// key: &str,
|
||||
// ) -> Result<Vec<u8>, String> {
|
||||
// let res = redis::cmd("GET").arg(key).query_async(&mut conn).await;
|
||||
|
||||
match res {
|
||||
Ok(res) => Ok(res),
|
||||
Err(e) => {
|
||||
return Err(format!("{e}"));
|
||||
}
|
||||
}
|
||||
}
|
||||
// match res {
|
||||
// Ok(res) => Ok(res),
|
||||
// Err(e) => {
|
||||
// return Err(format!("{e}"));
|
||||
// }
|
||||
// }
|
||||
// }
|
||||
|
||||
fn get_git_object(
|
||||
repo: MutexGuard<'_, Repository>,
|
||||
|
|
@ -675,97 +605,150 @@ fn get_git_object(
|
|||
}
|
||||
}
|
||||
|
||||
fn read_patch_from_str(patch_changes: &str) -> Option<Value> {
|
||||
//
|
||||
|
||||
let pv = patch_changes.replace("patch ", "");
|
||||
|
||||
let change_map: Value = match serde_json::from_str(&pv) {
|
||||
Ok(s) => s,
|
||||
Err(_) => return None,
|
||||
};
|
||||
Some(change_map)
|
||||
#[derive(Deserialize, Serialize, Debug)]
|
||||
struct SimplifiedCommit {
|
||||
oid: String,
|
||||
epoch_time_sec: i64,
|
||||
msg: String,
|
||||
author: (String, String),
|
||||
}
|
||||
|
||||
fn get_product_code(v: &Value) -> Option<&str> {
|
||||
v.get("productCode")?.as_str()
|
||||
}
|
||||
fn get_object_log(
|
||||
repo: MutexGuard<'_, Repository>,
|
||||
path: &str,
|
||||
) -> Result<Vec<SimplifiedCommit>, Box<dyn std::error::Error + Send + Sync>> {
|
||||
let mut walk = repo.revwalk()?;
|
||||
walk.push_head()?;
|
||||
walk.set_sorting(git2::Sort::TIME)?;
|
||||
|
||||
fn diff_apply(target: &mut Value, patch: &Value) -> Result<(), String> {
|
||||
match (target, patch) {
|
||||
(Value::Object(target_map), Value::Object(patch_map)) => {
|
||||
for (k, v_patch) in patch_map {
|
||||
let v_target = target_map
|
||||
.get_mut(k)
|
||||
.ok_or_else(|| format!("Unknown key in patch: {k}"))?;
|
||||
walk.simplify_first_parent()?;
|
||||
let walker: Vec<git2::Commit<'_>> = walk
|
||||
.filter_map(|oid| {
|
||||
let commit = repo.find_commit(oid.unwrap()).unwrap();
|
||||
let tree = commit.tree().unwrap();
|
||||
|
||||
diff_apply(v_target, v_patch)?;
|
||||
if let Ok(_) = tree.get_path(Path::new(path)) {
|
||||
Some(commit.clone())
|
||||
} else {
|
||||
None
|
||||
}
|
||||
})
|
||||
.collect();
|
||||
|
||||
Ok(())
|
||||
}
|
||||
(Value::Array(target_arr), Value::Array(patch_arr)) => {
|
||||
for patch_elem in patch_arr {
|
||||
// NOTE: support only `Recipe01`
|
||||
let patch_id = get_product_code(patch_elem)
|
||||
.ok_or("Patch array element missing product code")?;
|
||||
let target_elem = target_arr
|
||||
.iter_mut()
|
||||
.find(|e| get_product_code(e) == Some(patch_id))
|
||||
.ok_or_else(|| format!("Unknown id in patch array: {patch_id}"))?;
|
||||
let logs = walker
|
||||
.iter()
|
||||
.map(|c| {
|
||||
let cc = c.clone();
|
||||
|
||||
diff_apply(target_elem, patch_elem)?;
|
||||
let time = cc.time().seconds();
|
||||
let msg = cc.message().unwrap_or_default().to_string();
|
||||
|
||||
SimplifiedCommit {
|
||||
oid: cc.id().to_string(),
|
||||
epoch_time_sec: time,
|
||||
author: (
|
||||
cc.author().name().unwrap_or_default().to_string(),
|
||||
cc.author().email().unwrap_or_default().to_string(),
|
||||
),
|
||||
msg,
|
||||
}
|
||||
})
|
||||
.collect();
|
||||
|
||||
Ok(())
|
||||
}
|
||||
(target_slot, patch_value) => {
|
||||
*target_slot = patch_value.clone();
|
||||
Ok(())
|
||||
}
|
||||
}
|
||||
Ok(logs)
|
||||
}
|
||||
|
||||
async fn apply_patch_to_file(redis: RedisPool, path: &str, patch_changes: &mut String) -> String {
|
||||
use libtbr::*;
|
||||
// expect the path to already has in redis
|
||||
let full_file = match fetch_content_from_redis(redis, path).await {
|
||||
Ok(f) => f,
|
||||
Err(_) => String::new(),
|
||||
};
|
||||
// fn read_patch_from_str(patch_changes: &str) -> Option<Value> {
|
||||
// //
|
||||
|
||||
if full_file.is_empty() {
|
||||
return full_file;
|
||||
}
|
||||
// let pv = patch_changes.replace("patch ", "");
|
||||
|
||||
// read into struct
|
||||
//
|
||||
let full_recipe: models::recipe::Recipe = match serde_json::from_str(&full_file) {
|
||||
Ok(f) => f,
|
||||
Err(_) => return String::new(),
|
||||
};
|
||||
// let change_map: Value = match serde_json::from_str(&pv) {
|
||||
// Ok(s) => s,
|
||||
// Err(_) => return None,
|
||||
// };
|
||||
// Some(change_map)
|
||||
// }
|
||||
|
||||
// read patch
|
||||
let patch_map = read_patch_from_str(patch_changes);
|
||||
let mut current_full_map = match serde_json::to_value(full_recipe.clone()) {
|
||||
Ok(m) => m,
|
||||
Err(_) => return String::new(),
|
||||
};
|
||||
// fn get_product_code(v: &Value) -> Option<&str> {
|
||||
// v.get("productCode")?.as_str()
|
||||
// }
|
||||
|
||||
if let Some(pm) = patch_map {
|
||||
match diff_apply(&mut current_full_map, &pm) {
|
||||
Ok(_) => {}
|
||||
Err(x) => {
|
||||
error!("error while applied patch: {x}");
|
||||
}
|
||||
}
|
||||
}
|
||||
// fn diff_apply(target: &mut Value, patch: &Value) -> Result<(), String> {
|
||||
// match (target, patch) {
|
||||
// (Value::Object(target_map), Value::Object(patch_map)) => {
|
||||
// for (k, v_patch) in patch_map {
|
||||
// let v_target = target_map
|
||||
// .get_mut(k)
|
||||
// .ok_or_else(|| format!("Unknown key in patch: {k}"))?;
|
||||
|
||||
match serde_json::to_string_pretty(¤t_full_map.clone()) {
|
||||
Ok(ss) => ss,
|
||||
Err(_) => String::new(),
|
||||
}
|
||||
}
|
||||
// diff_apply(v_target, v_patch)?;
|
||||
// }
|
||||
|
||||
// Ok(())
|
||||
// }
|
||||
// (Value::Array(target_arr), Value::Array(patch_arr)) => {
|
||||
// for patch_elem in patch_arr {
|
||||
// // NOTE: support only `Recipe01`
|
||||
// let patch_id = get_product_code(patch_elem)
|
||||
// .ok_or("Patch array element missing product code")?;
|
||||
// let target_elem = target_arr
|
||||
// .iter_mut()
|
||||
// .find(|e| get_product_code(e) == Some(patch_id))
|
||||
// .ok_or_else(|| format!("Unknown id in patch array: {patch_id}"))?;
|
||||
|
||||
// diff_apply(target_elem, patch_elem)?;
|
||||
// }
|
||||
|
||||
// Ok(())
|
||||
// }
|
||||
// (target_slot, patch_value) => {
|
||||
// *target_slot = patch_value.clone();
|
||||
// Ok(())
|
||||
// }
|
||||
// }
|
||||
// }
|
||||
|
||||
// async fn apply_patch_to_file(redis: RedisPool, path: &str, patch_changes: &mut String) -> String {
|
||||
// use libtbr::*;
|
||||
// // expect the path to already has in redis
|
||||
// let full_file = match fetch_content_from_redis(redis, path).await {
|
||||
// Ok(f) => f,
|
||||
// Err(_) => String::new(),
|
||||
// };
|
||||
|
||||
// if full_file.is_empty() {
|
||||
// return full_file;
|
||||
// }
|
||||
|
||||
// // read into struct
|
||||
// //
|
||||
// let full_recipe: models::recipe::Recipe = match serde_json::from_str(&full_file) {
|
||||
// Ok(f) => f,
|
||||
// Err(_) => return String::new(),
|
||||
// };
|
||||
|
||||
// // read patch
|
||||
// let patch_map = read_patch_from_str(patch_changes);
|
||||
// let mut current_full_map = match serde_json::to_value(full_recipe.clone()) {
|
||||
// Ok(m) => m,
|
||||
// Err(_) => return String::new(),
|
||||
// };
|
||||
|
||||
// if let Some(pm) = patch_map {
|
||||
// match diff_apply(&mut current_full_map, &pm) {
|
||||
// Ok(_) => {}
|
||||
// Err(x) => {
|
||||
// error!("error while applied patch: {x}");
|
||||
// }
|
||||
// }
|
||||
// }
|
||||
|
||||
// match serde_json::to_string_pretty(¤t_full_map.clone()) {
|
||||
// Ok(ss) => ss,
|
||||
// Err(_) => String::new(),
|
||||
// }
|
||||
// }
|
||||
|
||||
async fn commit_file_content(
|
||||
repo: Arc<Mutex<Repository>>,
|
||||
|
|
@ -834,7 +817,7 @@ async fn commit_file_content(
|
|||
|
||||
repo_guard.reference(&target_ref, oid, true, "api commit log update")?;
|
||||
|
||||
if let Ok(mut head) = repo_guard.head() {
|
||||
if let Ok(head) = repo_guard.head() {
|
||||
if head.name() == Some(&target_ref) {
|
||||
repo_guard.set_head(&target_ref)?;
|
||||
}
|
||||
|
|
@ -1176,6 +1159,129 @@ async fn pull(
|
|||
Ok(())
|
||||
}
|
||||
|
||||
#[derive(Debug, Deserialize)]
|
||||
struct GetHistoryParams {
|
||||
path: String,
|
||||
from: String,
|
||||
to: Option<String>,
|
||||
}
|
||||
|
||||
/// Get history of file with date threshold filter
|
||||
///
|
||||
/// Query:
|
||||
/// - path: String , a path to the target file
|
||||
/// - from: String , a specified date for start, expected timestamp format
|
||||
/// - to (optional) : String , a date to stop filter, expected timestamp format
|
||||
/// default is current date timestamp.
|
||||
///
|
||||
/// Example:
|
||||
///
|
||||
/// ```
|
||||
/// GET /history?path=tha/latest_release_build_date&from=
|
||||
/// ```
|
||||
async fn get_history_handler(
|
||||
State(state): State<AppState>,
|
||||
Query(param): Query<GetHistoryParams>,
|
||||
) -> impl IntoResponse {
|
||||
let target_path = param.path;
|
||||
let filter_from = match param.from.parse::<i64>() {
|
||||
Ok(fi) => match chrono::DateTime::from_timestamp_secs(fi) {
|
||||
Some(mfi) => mfi,
|
||||
None => {
|
||||
return (
|
||||
axum::http::StatusCode::INTERNAL_SERVER_ERROR,
|
||||
"Cannot convert `from` to timestamp",
|
||||
)
|
||||
.into_response();
|
||||
}
|
||||
},
|
||||
Err(_) => {
|
||||
return (
|
||||
axum::http::StatusCode::BAD_REQUEST,
|
||||
"Not expected format on from",
|
||||
)
|
||||
.into_response();
|
||||
}
|
||||
};
|
||||
|
||||
let filter_stop = if let Some(ef) = param.to {
|
||||
match ef.parse::<i64>() {
|
||||
Ok(fi) => match chrono::DateTime::from_timestamp_secs(fi) {
|
||||
Some(mfi) => mfi,
|
||||
None => {
|
||||
return (
|
||||
axum::http::StatusCode::INTERNAL_SERVER_ERROR,
|
||||
"Cannot convert `to` to timestamp",
|
||||
)
|
||||
.into_response();
|
||||
}
|
||||
},
|
||||
Err(_) => {
|
||||
return (
|
||||
axum::http::StatusCode::BAD_REQUEST,
|
||||
"Not expected format on to",
|
||||
)
|
||||
.into_response();
|
||||
}
|
||||
}
|
||||
} else {
|
||||
chrono::Utc::now()
|
||||
};
|
||||
|
||||
// let fpath = format!(
|
||||
// "{}:{}",
|
||||
// state
|
||||
// .get_config("GIT_REPO_BRANCH_NAME")
|
||||
// .map(|x| x.to_string())
|
||||
// .unwrap_or("master".to_string()),
|
||||
// target_path
|
||||
// );
|
||||
let fpath = target_path;
|
||||
let repo = state.repo.clone();
|
||||
let fpath_clone = fpath.clone();
|
||||
|
||||
let result = tokio::task::spawn_blocking(move || {
|
||||
let repo = repo.try_lock()?;
|
||||
|
||||
get_object_log(repo, &fpath_clone)
|
||||
})
|
||||
.await;
|
||||
|
||||
match result {
|
||||
Ok(Ok(o)) => {
|
||||
// filter out
|
||||
let f: Vec<String> = o
|
||||
.iter()
|
||||
.filter_map(|oc| {
|
||||
let occ = oc;
|
||||
if let Some(t) = chrono::DateTime::from_timestamp_secs(occ.epoch_time_sec)
|
||||
&& t >= filter_from
|
||||
&& t <= filter_stop
|
||||
{
|
||||
Some(format!(
|
||||
"{}\t{}\t{}\t{}, ({})",
|
||||
t.to_rfc3339(),
|
||||
occ.msg,
|
||||
occ.oid,
|
||||
occ.author.0,
|
||||
occ.author.1
|
||||
))
|
||||
} else {
|
||||
None
|
||||
}
|
||||
})
|
||||
.collect();
|
||||
|
||||
(axum::http::StatusCode::OK, serde_json::json!(f).to_string()).into_response()
|
||||
}
|
||||
e => {
|
||||
error!("error while fetching history: {e:?}");
|
||||
|
||||
(axum::http::StatusCode::NOT_FOUND, "Not found").into_response()
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// Return git state & server state
|
||||
async fn health_handler(State(_): State<AppState>) -> impl IntoResponse {
|
||||
(
|
||||
|
|
@ -1207,7 +1313,7 @@ pub async fn run(config: gcm::Configure) -> gcm::StandardResult {
|
|||
.get("POSTGRES_CONN")
|
||||
.expect("failed to connect postgres");
|
||||
info!("connect config: {connect_str}");
|
||||
let (mut client, connection) = tokio_postgres::connect(connect_str, NoTls).await?;
|
||||
let (client, connection) = tokio_postgres::connect(connect_str, NoTls).await?;
|
||||
tokio::spawn(async move {
|
||||
if let Err(e) = connection.await {
|
||||
error!("connection postgres error: {e}");
|
||||
|
|
@ -1237,8 +1343,8 @@ pub async fn run(config: gcm::Configure) -> gcm::StandardResult {
|
|||
config.get("GIT_REPO_LOCAL_DEST").unwrap_or(&"".to_string()),
|
||||
)?)),
|
||||
postgres: Arc::new(Mutex::new(client)),
|
||||
redis: redis_pool.clone(),
|
||||
fetched: Vec::new(),
|
||||
_redis: redis_pool.clone(),
|
||||
_fetched: Vec::new(),
|
||||
};
|
||||
|
||||
let app = Router::new()
|
||||
|
|
@ -1249,6 +1355,7 @@ pub async fn run(config: gcm::Configure) -> gcm::StandardResult {
|
|||
.route("/pull", get(pull_handler))
|
||||
.route("/health", get(health_handler))
|
||||
.route("/replay", post(replay_handler))
|
||||
.route("/history", get(get_history_handler))
|
||||
.layer(DefaultBodyLimit::max(100 * 1024 * 1024))
|
||||
// .route("/healthz", get(reg::health))
|
||||
.with_state(state);
|
||||
|
|
|
|||
|
|
@ -1,5 +1,3 @@
|
|||
use std::fs::File;
|
||||
|
||||
use env_logger::Builder;
|
||||
use libtbr::recipe_functions::common;
|
||||
use log::info;
|
||||
|
|
@ -12,10 +10,9 @@ use crate::{
|
|||
mod app;
|
||||
mod gcm;
|
||||
mod git;
|
||||
mod tx;
|
||||
// mod reg;
|
||||
|
||||
fn setup_log(config: gcm::Configure) -> gcm::StandardResult {
|
||||
fn setup_log(_config: gcm::Configure) -> gcm::StandardResult {
|
||||
// NOTE: disable logging file, use send to log service instead
|
||||
// let logfile = File::create(config.get("LOG_NAME").unwrap_or(&"run.log".to_string()))?;
|
||||
|
||||
|
|
|
|||
|
|
@ -1,323 +0,0 @@
|
|||
use axum::{
|
||||
Json, Router,
|
||||
extract::{Path, Query, State},
|
||||
response::IntoResponse,
|
||||
routing::{get, post},
|
||||
};
|
||||
use redis::{AsyncTypedCommands, TypedCommands, aio::MultiplexedConnection};
|
||||
use reqwest::StatusCode;
|
||||
use serde::Deserialize;
|
||||
use serde_json::{Value, json};
|
||||
use uuid::Uuid;
|
||||
|
||||
use crate::tx::{
|
||||
helpers::*,
|
||||
patcher::apply_ops,
|
||||
types::{PatchOp, Reservation, ReserveReq, ReserveRes, TxCommitReq, TxCommitRes},
|
||||
};
|
||||
|
||||
async fn reserve_tx(
|
||||
State(state): State<AppState>,
|
||||
Path(doc_id): Path<String>,
|
||||
Json(req): Json<ReserveReq>,
|
||||
) -> impl IntoResponse {
|
||||
let rsv_id = Uuid::new_v4().to_string();
|
||||
let ttl_secs: i64 = 30; // ttl reserve 30s
|
||||
let expires_at = now_unix() + ttl_secs;
|
||||
|
||||
let mut rcli = state.redis_cli.clone();
|
||||
|
||||
// check version
|
||||
let ver_key = doc_ver_key(&doc_id);
|
||||
let base_version: u64 = match rcli.get(ver_key) {
|
||||
Ok(res) => {
|
||||
if let Some(r) = res {
|
||||
r.parse().ok().unwrap()
|
||||
} else {
|
||||
0
|
||||
}
|
||||
}
|
||||
Err(_) => 0,
|
||||
};
|
||||
|
||||
let rsv = Reservation {
|
||||
reservation_id: rsv_id.clone(),
|
||||
doc_id: doc_id.clone(),
|
||||
author: req.author,
|
||||
base_version,
|
||||
expires_at,
|
||||
};
|
||||
|
||||
let key = rsv_key(&doc_id, &rsv_id);
|
||||
let payload = serde_json::to_string(&rsv).unwrap();
|
||||
|
||||
let _: () = rcli.set_ex(key, payload, ttl_secs as u64).unwrap();
|
||||
|
||||
(
|
||||
StatusCode::OK,
|
||||
Json(ReserveRes {
|
||||
reservation_id: rsv_id,
|
||||
base_version,
|
||||
expires_at,
|
||||
}),
|
||||
)
|
||||
}
|
||||
|
||||
async fn commit_tx(
|
||||
State(state): State<AppState>,
|
||||
Path(doc_id): Path<String>,
|
||||
Json(req): Json<TxCommitReq>,
|
||||
) -> impl IntoResponse {
|
||||
if req.tx_id.is_empty() || req.reservation_id.is_empty() {
|
||||
return (
|
||||
StatusCode::BAD_REQUEST,
|
||||
"missing tx_id or reservation_id ".into_response(),
|
||||
);
|
||||
}
|
||||
|
||||
let mut con = match state.redis_cli.get_multiplexed_async_connection().await {
|
||||
Ok(conn) => conn,
|
||||
Err(e) => {
|
||||
return (
|
||||
StatusCode::INTERNAL_SERVER_ERROR,
|
||||
format!("{}", e.to_string()).into_response(),
|
||||
);
|
||||
}
|
||||
};
|
||||
|
||||
// lock commit
|
||||
let lock_key = commit_lock_key(&doc_id);
|
||||
if let Err(_) = wait_lock(&mut con, &lock_key, 1200, 20).await {
|
||||
return (StatusCode::CONFLICT, "doc busy, retry".into_response());
|
||||
}
|
||||
|
||||
let txid_k = txid_key(&doc_id, &req.tx_id);
|
||||
if let Ok(Some(v)) = con.get(&txid_k).await {
|
||||
if let Ok(committed_version) = v.parse::<u64>() {
|
||||
let res = TxCommitRes {
|
||||
doc_id,
|
||||
tx_id: req.tx_id,
|
||||
committed_version,
|
||||
};
|
||||
return (StatusCode::OK, Json(res).into_response());
|
||||
}
|
||||
}
|
||||
|
||||
let rsv_k = rsv_key(&doc_id, &req.reservation_id);
|
||||
let rsv_raw: Option<String> = con.get(&rsv_k).await.unwrap_or(None);
|
||||
let rsv_raw = match rsv_raw {
|
||||
Some(x) => x,
|
||||
None => {
|
||||
return (
|
||||
StatusCode::CONFLICT,
|
||||
"reservation missing/expired".into_response(),
|
||||
);
|
||||
}
|
||||
};
|
||||
|
||||
let rsv: Reservation = match serde_json::from_str(&rsv_raw) {
|
||||
Ok(x) => x,
|
||||
Err(_) => return (StatusCode::CONFLICT, "bad reservation".into_response()),
|
||||
};
|
||||
|
||||
if rsv.doc_id != doc_id || rsv.author != req.author || rsv.base_version != req.base_version {
|
||||
return (StatusCode::CONFLICT, "reservation mismatch".into_response());
|
||||
}
|
||||
|
||||
if now_unix() > rsv.expires_at {
|
||||
return (StatusCode::CONFLICT, "reservation expired".into_response());
|
||||
}
|
||||
|
||||
let curr_ver: u64 = con
|
||||
.get(doc_ver_key(&doc_id))
|
||||
.await
|
||||
.ok()
|
||||
.flatten()
|
||||
.and_then(|s| s.parse().ok())
|
||||
.unwrap_or(0);
|
||||
|
||||
if req.base_version != curr_ver {
|
||||
let body = json!({
|
||||
"error": "version_conflict",
|
||||
"server_version": curr_ver,
|
||||
});
|
||||
return (StatusCode::CONFLICT, Json(body).into_response());
|
||||
}
|
||||
|
||||
let next_ver = curr_ver + 1;
|
||||
|
||||
let stream_key = doc_tx_stream_key(&doc_id);
|
||||
|
||||
let tx_event = json!({
|
||||
"doc_id": doc_id,
|
||||
"tx_id": req.tx_id,
|
||||
"version": next_ver,
|
||||
"base_version": req.base_version,
|
||||
"author": req.author,
|
||||
"ts": now_unix(),
|
||||
"ops": req.ops
|
||||
});
|
||||
|
||||
let tx_event_str = tx_event.to_string();
|
||||
let _: String = redis::cmd("XADD")
|
||||
.arg(&stream_key)
|
||||
.arg("*")
|
||||
.arg("data")
|
||||
.arg(&tx_event_str)
|
||||
.query_async(&mut con)
|
||||
.await
|
||||
.unwrap();
|
||||
|
||||
let _: () = con
|
||||
.set(doc_ver_key(&doc_id), next_ver.to_string())
|
||||
.await
|
||||
.unwrap();
|
||||
let _: () = con.set(&txid_k, next_ver.to_string()).await.unwrap();
|
||||
let _: usize = con.del(&rsv_k).await.unwrap();
|
||||
|
||||
let snapshot_every: u64 = 50;
|
||||
if next_ver % snapshot_every == 0 {
|
||||
if let Ok(state) = build_state_at(&mut con, &doc_id, next_ver).await {
|
||||
let _: () = con
|
||||
.set(doc_snap_key(&doc_id), state.to_string())
|
||||
.await
|
||||
.unwrap();
|
||||
let _: () = con
|
||||
.set(doc_snapver_key(&doc_id), next_ver.to_string())
|
||||
.await
|
||||
.unwrap();
|
||||
}
|
||||
}
|
||||
|
||||
(
|
||||
StatusCode::OK,
|
||||
Json(TxCommitRes {
|
||||
doc_id,
|
||||
tx_id: req.tx_id,
|
||||
committed_version: next_ver,
|
||||
})
|
||||
.into_response(),
|
||||
)
|
||||
}
|
||||
|
||||
async fn build_state_at(
|
||||
con: &mut MultiplexedConnection,
|
||||
doc_id: &str,
|
||||
target_version: u64,
|
||||
) -> Result<Value, String> {
|
||||
let snap_ver: u64 = con
|
||||
.get(doc_snap_key(&doc_id))
|
||||
.await
|
||||
.ok()
|
||||
.flatten()
|
||||
.and_then(|s| s.parse().ok())
|
||||
.unwrap_or(0);
|
||||
|
||||
let snap_json: Value = if snap_ver > 0 {
|
||||
let raw: Option<String> = con.get(doc_snap_key(&doc_id)).await.unwrap_or(None);
|
||||
raw.and_then(|s| serde_json::from_str(&s).ok())
|
||||
.unwrap_or_else(|| json!({}))
|
||||
} else {
|
||||
json!({})
|
||||
};
|
||||
|
||||
let stream_key = doc_tx_stream_key(doc_id);
|
||||
|
||||
let start_id = "0-0";
|
||||
let entries: Vec<(String, Vec<(String, String)>)> = redis::cmd("XRANGE")
|
||||
.arg(&stream_key)
|
||||
.arg(start_id)
|
||||
.arg("+")
|
||||
.query_async(con)
|
||||
.await
|
||||
.map_err(|e| e.to_string())?;
|
||||
|
||||
let mut ops_list: Vec<PatchOp> = Vec::new();
|
||||
for (_id, fields) in entries {
|
||||
let mut data_opts = None;
|
||||
for (k, v) in fields {
|
||||
if k == "data" {
|
||||
data_opts = Some(v);
|
||||
break;
|
||||
}
|
||||
}
|
||||
let data = match data_opts {
|
||||
Some(x) => x,
|
||||
None => continue,
|
||||
};
|
||||
|
||||
let ev: Value = serde_json::from_str(&data).map_err(|e| e.to_string())?;
|
||||
let ver = ev.get("version").and_then(|v| v.as_u64()).unwrap_or(0);
|
||||
if ver == 0 || ver > target_version {
|
||||
continue;
|
||||
}
|
||||
|
||||
if ver <= snap_ver {
|
||||
continue;
|
||||
}
|
||||
|
||||
let ops: Vec<PatchOp> =
|
||||
serde_json::from_value(ev["ops"].clone()).map_err(|e| e.to_string())?;
|
||||
ops_list.extend(ops);
|
||||
}
|
||||
|
||||
apply_ops(snap_json, &ops_list)
|
||||
}
|
||||
|
||||
#[derive(Deserialize)]
|
||||
struct StateQuery {
|
||||
at_version: Option<u64>,
|
||||
}
|
||||
|
||||
async fn get_state(
|
||||
State(state): State<AppState>,
|
||||
Path(doc_id): Path<String>,
|
||||
Query(q): Query<StateQuery>,
|
||||
) -> impl IntoResponse {
|
||||
let rcli = state.redis_cli.clone();
|
||||
let mut con = match rcli.get_multiplexed_async_connection().await {
|
||||
Ok(sc) => sc,
|
||||
Err(_) => {
|
||||
return (
|
||||
StatusCode::INTERNAL_SERVER_ERROR,
|
||||
"cannot get connection".into_response(),
|
||||
);
|
||||
}
|
||||
};
|
||||
|
||||
let cur_ver: u64 = con
|
||||
.get(doc_ver_key(&doc_id))
|
||||
.await
|
||||
.ok()
|
||||
.flatten()
|
||||
.and_then(|s| s.parse().ok())
|
||||
.unwrap_or(0);
|
||||
|
||||
let target = q.at_version.unwrap_or(cur_ver).min(cur_ver);
|
||||
|
||||
match build_state_at(&mut con, &doc_id, target).await {
|
||||
Ok(state) => (
|
||||
StatusCode::OK,
|
||||
Json(json!({
|
||||
"doc_id": doc_id,
|
||||
"version": target,
|
||||
"state": state
|
||||
}))
|
||||
.into_response(),
|
||||
),
|
||||
Err(e) => (
|
||||
StatusCode::INTERNAL_SERVER_ERROR,
|
||||
Json(json!({
|
||||
"error": e
|
||||
}))
|
||||
.into_response(),
|
||||
),
|
||||
}
|
||||
}
|
||||
|
||||
// pub async fn create_tx_patcher_route() -> Router<AppState> {
|
||||
// Router::new()
|
||||
// .route("/{id}/tx-reserve", post(reserve_tx))
|
||||
// .route("/{id}/tx-commit", post(commit_tx))
|
||||
// .route("/{id}/state", get(get_state))
|
||||
// }
|
||||
|
|
@ -1,296 +0,0 @@
|
|||
use std::{
|
||||
clone,
|
||||
collections::HashMap,
|
||||
sync::{Arc, Mutex},
|
||||
time::{Duration, SystemTime, UNIX_EPOCH},
|
||||
};
|
||||
|
||||
use git2::Repository;
|
||||
use json_patch::{Patch, diff};
|
||||
use log::info;
|
||||
use redis::{RedisResult, aio::MultiplexedConnection};
|
||||
use serde::Deserialize;
|
||||
use tokio::{sync::mpsc::Sender, time::sleep};
|
||||
use uuid::Uuid;
|
||||
|
||||
use crate::app::CacheJob;
|
||||
|
||||
pub fn now_unix() -> i64 {
|
||||
SystemTime::now()
|
||||
.duration_since(UNIX_EPOCH)
|
||||
.unwrap()
|
||||
.as_secs() as i64
|
||||
}
|
||||
|
||||
pub fn doc_ver_key(doc: &str) -> String {
|
||||
format!("doc:{doc}:ver")
|
||||
}
|
||||
pub fn doc_tx_stream_key(doc: &str) -> String {
|
||||
format!("doc:{doc}:tx")
|
||||
}
|
||||
pub fn doc_snap_key(doc: &str) -> String {
|
||||
format!("doc:{doc}:snap")
|
||||
}
|
||||
pub fn doc_snapver_key(doc: &str) -> String {
|
||||
format!("doc:{doc}:snapver")
|
||||
}
|
||||
pub fn rsv_key(doc: &str, rsv: &str) -> String {
|
||||
format!("doc:{doc}:rsv:{rsv}")
|
||||
}
|
||||
pub fn txid_key(doc: &str, tx_id: &str) -> String {
|
||||
format!("doc:{doc}:txid:{tx_id}")
|
||||
}
|
||||
pub fn commit_lock_key(doc: &str) -> String {
|
||||
format!("doc:{doc}:lock")
|
||||
}
|
||||
|
||||
pub async fn acquire_lock(
|
||||
con: &mut MultiplexedConnection,
|
||||
key: &str,
|
||||
ttl_ms: u64,
|
||||
) -> RedisResult<bool> {
|
||||
let token = Uuid::new_v4().to_string();
|
||||
let ok: Option<String> = redis::cmd("SET")
|
||||
.arg(key)
|
||||
.arg(token)
|
||||
.arg("NX")
|
||||
.arg("PX")
|
||||
.arg(ttl_ms)
|
||||
.query_async(con)
|
||||
.await?;
|
||||
Ok(ok.is_some())
|
||||
}
|
||||
|
||||
pub async fn wait_lock(
|
||||
con: &mut MultiplexedConnection,
|
||||
key: &str,
|
||||
ttl_ms: u64,
|
||||
attempts: u32,
|
||||
) -> RedisResult<()> {
|
||||
for _ in 0..attempts {
|
||||
if acquire_lock(con, key, ttl_ms).await? {
|
||||
return Ok(());
|
||||
}
|
||||
|
||||
sleep(Duration::from_millis(30)).await;
|
||||
}
|
||||
Err(redis::RedisError::from((redis::ErrorKind::Io, "lock busy")))
|
||||
}
|
||||
|
||||
pub fn build_tx_from_diff(base: serde_json::Value, alt: serde_json::Value) -> Patch {
|
||||
diff(&base, &alt)
|
||||
}
|
||||
|
||||
fn get_file_from_repo(
|
||||
repo: Arc<Mutex<Repository>>,
|
||||
path: &str,
|
||||
) -> Result<String, Box<dyn std::error::Error + Send + Sync>> {
|
||||
if let Ok(repo_m) = repo.try_lock() {
|
||||
let obj = repo_m.revparse_single(path)?;
|
||||
if let Some(blob) = obj.as_blob() {
|
||||
let content = unsafe { str::from_utf8_unchecked(blob.content()) };
|
||||
return Ok(content.to_string());
|
||||
} else if let Some(tree) = obj.as_tree() {
|
||||
let dir_list = tree
|
||||
.iter()
|
||||
.map(|x| x.name().unwrap_or("").to_string())
|
||||
.collect::<Vec<String>>();
|
||||
|
||||
return Ok(dir_list.join(","));
|
||||
}
|
||||
}
|
||||
Err("FileNotFound".into())
|
||||
}
|
||||
|
||||
pub async fn setup_prebuild_tx_cache(
|
||||
tx: Sender<CacheJob>,
|
||||
git_repo: &str,
|
||||
) -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
|
||||
// import tx files
|
||||
let repo = match Repository::open(git_repo) {
|
||||
Ok(repo) => repo,
|
||||
Err(_) => return Err("cannot open repo".into()),
|
||||
};
|
||||
|
||||
let repo_m = Arc::new(Mutex::new(repo));
|
||||
|
||||
let root_path = String::from("master:");
|
||||
let all_folders = match get_file_from_repo(repo_m.clone(), root_path.as_str()) {
|
||||
Ok(af) => af,
|
||||
Err(e) => return Err(e),
|
||||
};
|
||||
|
||||
let folders = all_folders
|
||||
.split(",")
|
||||
.map(|x| x.to_string())
|
||||
.collect::<Vec<String>>();
|
||||
|
||||
for folder in folders {
|
||||
let current_path = format!("master:{folder}");
|
||||
|
||||
println!("trying {folder}...");
|
||||
|
||||
// test get version
|
||||
//
|
||||
// NOTE: version may miss if there is file with `versionXXXX`
|
||||
//
|
||||
//
|
||||
if let Ok(current_files) = get_file_from_repo(repo_m.clone(), current_path.as_str())
|
||||
&& current_files.contains(",version")
|
||||
{
|
||||
let expected_version_path = format!("{current_path}/version");
|
||||
println!("\tmay have version --> {expected_version_path}, \ncurrent: {current_files}");
|
||||
|
||||
// get version content
|
||||
|
||||
let version_str =
|
||||
if let Ok(v) = get_file_from_repo(repo_m.clone(), expected_version_path.as_str()) {
|
||||
v
|
||||
} else {
|
||||
"".to_string()
|
||||
};
|
||||
// let version = version_str.parse::<u64>().unwrap_or(0);
|
||||
if version_str.is_empty() {
|
||||
// unexpected
|
||||
return Err("cannot get version".into());
|
||||
}
|
||||
|
||||
if tx
|
||||
.send(CacheJob {
|
||||
rel_path: expected_version_path.clone(),
|
||||
file_data: Some(serde_json::to_vec(&version_str)?),
|
||||
})
|
||||
.await
|
||||
.is_ok()
|
||||
{
|
||||
info!("cache ok {current_path}/version");
|
||||
}
|
||||
|
||||
println!("try get {folder}:{version_str}");
|
||||
|
||||
let files = current_files
|
||||
.split(",")
|
||||
.map(|x| x.to_string())
|
||||
.collect::<Vec<String>>();
|
||||
|
||||
let base_file: String = match files.iter().find(|x| {
|
||||
x.ends_with(".json") && x.starts_with(&format!("coffeethai02_{version_str}"))
|
||||
}) {
|
||||
Some(f) => f.to_string(),
|
||||
None => "".to_string(),
|
||||
};
|
||||
|
||||
println!("{folder}: {base_file}");
|
||||
|
||||
if base_file.is_empty() {
|
||||
println!("base file empty skip");
|
||||
continue;
|
||||
}
|
||||
|
||||
let base_recipe_str = if let Ok(r) =
|
||||
get_file_from_repo(repo_m.clone(), &format!("{current_path}/{base_file}"))
|
||||
{
|
||||
r.to_string()
|
||||
} else {
|
||||
"".to_string()
|
||||
};
|
||||
|
||||
if base_recipe_str.is_empty() {
|
||||
println!("empty recipe skip!");
|
||||
continue;
|
||||
}
|
||||
|
||||
let base_recipe: serde_json::Value = match serde_json::from_str(&base_recipe_str) {
|
||||
Ok(r) => r,
|
||||
Err(e) => return Err(e.into()),
|
||||
};
|
||||
|
||||
let mut current_recipes = HashMap::new();
|
||||
|
||||
current_recipes.insert("base".to_string(), base_recipe);
|
||||
|
||||
for cfile in files {
|
||||
if cfile.ne(&base_file)
|
||||
&& cfile.ends_with(".json")
|
||||
&& cfile.starts_with("coffeethai02")
|
||||
{
|
||||
//
|
||||
// fetch file content
|
||||
// read into value
|
||||
// do diff
|
||||
|
||||
let current_fetch_path = format!("{current_path}/{cfile}");
|
||||
let current_file_str =
|
||||
if let Ok(cr) = get_file_from_repo(repo_m.clone(), ¤t_fetch_path) {
|
||||
cr.to_string()
|
||||
} else {
|
||||
"".to_string()
|
||||
};
|
||||
|
||||
if !current_file_str.is_empty() {
|
||||
// into value
|
||||
let alt_recipe: serde_json::Value =
|
||||
match serde_json::from_str(¤t_file_str) {
|
||||
Ok(crr) => crr,
|
||||
Err(e) => {
|
||||
println!("get error while alt {current_fetch_path}: {e}");
|
||||
continue;
|
||||
}
|
||||
};
|
||||
|
||||
current_recipes.insert(cfile, alt_recipe);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// process diffing
|
||||
//
|
||||
|
||||
let base_recipe_ref = current_recipes.get("base").unwrap();
|
||||
|
||||
for (key, value) in current_recipes.iter() {
|
||||
if key != "base" {
|
||||
let patch_diff = diff(&base_recipe_ref, value);
|
||||
let tx_name = format!(
|
||||
"stx_{}",
|
||||
key.replace("coffeethai02_", "").replace(".json", "")
|
||||
);
|
||||
|
||||
let base_path_out = format!("./recipe_tx/{folder}");
|
||||
if !std::path::Path::new(&base_path_out).exists() {
|
||||
std::fs::create_dir_all(&base_path_out)?;
|
||||
}
|
||||
|
||||
let file_out = format!("{base_path_out}/{tx_name}.json");
|
||||
|
||||
std::fs::write(file_out, serde_json::to_string_pretty(&patch_diff)?)?;
|
||||
}
|
||||
}
|
||||
|
||||
// let _ = tx
|
||||
// .send(serde_json::json!({
|
||||
// "topic": "generate_stx",
|
||||
// "target": folder,
|
||||
// "base_version": version_str,
|
||||
// "full_base_filename": base_file,
|
||||
// "value": base_recipe_clone
|
||||
// }))
|
||||
// .await;
|
||||
|
||||
let json_bytes = serde_json::to_vec(base_recipe_ref)?;
|
||||
|
||||
if tx
|
||||
.send(CacheJob {
|
||||
rel_path: format!("{current_path}/{base_file}"),
|
||||
file_data: Some(json_bytes),
|
||||
})
|
||||
.await
|
||||
.is_ok()
|
||||
{
|
||||
info!("cache ok {current_path}/{base_file}");
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
Ok(())
|
||||
}
|
||||
|
|
@ -1,4 +0,0 @@
|
|||
// pub mod handler;
|
||||
pub mod helpers;
|
||||
pub mod patcher;
|
||||
pub mod types;
|
||||
|
|
@ -1,119 +0,0 @@
|
|||
use serde_json::Value;
|
||||
|
||||
use crate::tx::types::PatchOp;
|
||||
|
||||
pub fn json_pointer_parent<'a>(
|
||||
root: &'a mut Value,
|
||||
path: &str,
|
||||
) -> Result<(&'a mut Value, String), String> {
|
||||
if !path.starts_with('/') && path != "" {
|
||||
return Err("path must be JSON pointer starting with '/'".into());
|
||||
}
|
||||
|
||||
let part: Vec<String> = path
|
||||
.split('/')
|
||||
.skip(1)
|
||||
.map(|x| x.replace("~1", "/").replace("~0", "~"))
|
||||
.collect();
|
||||
if path.is_empty() {
|
||||
return Ok((root, "".into()));
|
||||
}
|
||||
|
||||
let last = part.last().unwrap().clone();
|
||||
|
||||
let mut cur = root;
|
||||
|
||||
for key in &part[..part.len() - 1] {
|
||||
match cur {
|
||||
Value::Object(map) => {
|
||||
cur = map
|
||||
.get_mut(key)
|
||||
.ok_or_else(|| format!("missing object key {key}"))?;
|
||||
}
|
||||
Value::Array(arr) => {
|
||||
let idx: usize = key.parse().map_err(|_| format!("bad array idx {key}"))?;
|
||||
cur = arr
|
||||
.get_mut(idx)
|
||||
.ok_or_else(|| format!("array index out of bounds {idx}"))?;
|
||||
}
|
||||
_ => return Err("cannot traverse non-container".into()),
|
||||
}
|
||||
}
|
||||
|
||||
Ok((cur, last))
|
||||
}
|
||||
|
||||
pub fn apply_ops(mut state: Value, ops: &[PatchOp]) -> Result<Value, String> {
|
||||
for op in ops {
|
||||
match op {
|
||||
PatchOp::Replace { path, value } => {
|
||||
if path == "" || path == "/" {
|
||||
state = value.clone();
|
||||
continue;
|
||||
}
|
||||
let (parent, last) = json_pointer_parent(&mut state, path)?;
|
||||
match parent {
|
||||
Value::Object(map) => {
|
||||
if !map.contains_key(&last) {
|
||||
return Err(format!("replace target missing key {last}"));
|
||||
}
|
||||
map.insert(last, value.clone());
|
||||
}
|
||||
Value::Array(arr) => {
|
||||
let idx: usize = last
|
||||
.parse()
|
||||
.map_err(|_| format!("bad array index {last}"))?;
|
||||
if idx >= arr.len() {
|
||||
return Err(format!("replace index out of bounds {idx}"));
|
||||
}
|
||||
arr[idx] = value.clone();
|
||||
}
|
||||
_ => return Err("replace parent not container".into()),
|
||||
}
|
||||
}
|
||||
PatchOp::Add { path, value } => {
|
||||
let (parent, last) = json_pointer_parent(&mut state, path)?;
|
||||
match parent {
|
||||
Value::Object(map) => {
|
||||
map.insert(last, value.clone());
|
||||
}
|
||||
Value::Array(arr) => {
|
||||
if last == "-" {
|
||||
arr.push(value.clone());
|
||||
} else {
|
||||
let idx: usize = last
|
||||
.parse()
|
||||
.map_err(|_| format!("bad array index {last}"))?;
|
||||
if idx > arr.len() {
|
||||
return Err(format!("add index out of bounds {idx}"));
|
||||
}
|
||||
arr.insert(idx, value.clone());
|
||||
}
|
||||
}
|
||||
_ => return Err("add parent not container".into()),
|
||||
}
|
||||
}
|
||||
PatchOp::Remove { path } => {
|
||||
let (parent, last) = json_pointer_parent(&mut state, path)?;
|
||||
match parent {
|
||||
Value::Object(map) => {
|
||||
map.remove(&last)
|
||||
.ok_or_else(|| format!("remove missing key {last}"))?;
|
||||
}
|
||||
Value::Array(arr) => {
|
||||
let idx: usize = last
|
||||
.parse()
|
||||
.map_err(|_| format!("bad array index {last}"))?;
|
||||
if idx > arr.len() {
|
||||
return Err(format!("remove index out of bounds {idx}"));
|
||||
}
|
||||
arr.remove(idx);
|
||||
}
|
||||
_ => return Err("remove parent not container".into()),
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
Ok(state)
|
||||
}
|
||||
|
|
@ -1,51 +0,0 @@
|
|||
use serde::{Deserialize, Serialize};
|
||||
use serde_json::Value;
|
||||
|
||||
#[derive(Debug, Serialize, Deserialize)]
|
||||
pub struct ReserveReq {
|
||||
pub author: String,
|
||||
hint_paths: Option<Vec<String>>,
|
||||
}
|
||||
|
||||
#[derive(Debug, Serialize, Deserialize)]
|
||||
pub struct ReserveRes {
|
||||
pub reservation_id: String,
|
||||
pub base_version: u64,
|
||||
pub expires_at: i64,
|
||||
}
|
||||
|
||||
#[derive(Debug, Serialize, Deserialize, Clone)]
|
||||
pub struct TxCommitReq {
|
||||
pub reservation_id: String,
|
||||
pub tx_id: String,
|
||||
pub base_version: u64,
|
||||
pub author: String,
|
||||
pub ops: Vec<PatchOp>,
|
||||
}
|
||||
|
||||
#[derive(Debug, Serialize, Deserialize)]
|
||||
pub struct TxCommitRes {
|
||||
pub doc_id: String,
|
||||
pub tx_id: String,
|
||||
pub committed_version: u64,
|
||||
}
|
||||
|
||||
#[derive(Debug, Serialize, Deserialize, Clone)]
|
||||
#[serde(tag = "op")]
|
||||
pub enum PatchOp {
|
||||
#[serde(rename = "replace")]
|
||||
Replace { path: String, value: Value },
|
||||
#[serde(rename = "add")]
|
||||
Add { path: String, value: Value },
|
||||
#[serde(rename = "remove")]
|
||||
Remove { path: String },
|
||||
}
|
||||
|
||||
#[derive(Debug, Serialize, Deserialize)]
|
||||
pub struct Reservation {
|
||||
pub reservation_id: String,
|
||||
pub doc_id: String,
|
||||
pub author: String,
|
||||
pub base_version: u64,
|
||||
pub expires_at: i64,
|
||||
}
|
||||
Loading…
Add table
Add a link
Reference in a new issue