WEBDIENSTE/huly-platform
1
0
Fork 0
mirror of https://github.com/hcengineering/platform.git synced 2026-09-10 19:57:43 +02:00
Code Issues Packages Projects Releases Wiki Activity

implement conditional put/patch

Browse Source 
Signed-off-by: Alexander Onnikov <Alexander.Onnikov@xored.com>
This commit is contained in:
Alexander Onnikov
2025-09-22 23:41:20 +07:00
parent 2469c00b1d
commit 72dae37dd6
8 changed files with 542 additions and 48 deletions
Download Patch File Download Diff File Expand all files Collapse all files
server/src/conditional.rs
+167
View File
@@ -0,0 +1,167 @@
use actix_web::HttpRequest;
use actix_web::http::header::EntityTag;
use actix_web::http::header::Header;
use actix_web::http::header::IfMatch;
use actix_web::http::header::IfNoneMatch;
#[derive(thiserror::Error, Debug, PartialEq, Eq)]
pub enum ConditionalError {
#[error("Invalid header")]
ParseError,
}
#[derive(Debug, PartialEq, Eq)]
pub enum ConditionalMatch {
IfMatch(String),
IfNoneMatch(String),
}
pub fn any_match(
req: &HttpRequest,
etag: Option<EntityTag>,
) -> Result<Option<bool>, ConditionalError> {
match IfMatch::parse(req) {
Ok(IfMatch::Any) => Ok(Some(etag.is_some())),
Ok(IfMatch::Items(items)) => Ok({
if items.is_empty() {
None
} else {
Some(etag.map_or(false, |e| items.contains(&e)))
}
}),
Err(_) => Err(ConditionalError::ParseError),
}
}
pub fn none_match(
req: &HttpRequest,
etag: Option<EntityTag>,
) -> Result<Option<bool>, ConditionalError> {
match IfNoneMatch::parse(req) {
Ok(IfNoneMatch::Any) => Ok(Some(etag.is_none())),
Ok(IfNoneMatch::Items(items)) => Ok({
if items.is_empty() {
None
} else {
Some(etag.map_or(true, |e| !items.contains(&e)))
}
}),
Err(_) => Err(ConditionalError::ParseError),
}
}
#[cfg(test)]
mod tests {
use super::*;
use actix_web::test::TestRequest;
#[test]
fn test_any_match_none_some() {
assert_eq!(
any_match(
&TestRequest::default().to_http_request(),
Some(EntityTag::new_strong("foo".to_owned()))
),
Ok(None)
);
}
#[test]
fn test_any_match_any_some() {
assert_eq!(
any_match(
&TestRequest::default()
.insert_header(("If-Match", "*"))
.to_http_request(),
Some(EntityTag::new_strong("foo".to_owned()))
),
Ok(Some(true))
);
}
#[test]
fn test_any_match_some_some() {
assert_eq!(
any_match(
&TestRequest::default()
.insert_header(("If-Match", "\"foo\" ,\"bar\""))
.to_http_request(),
Some(EntityTag::new_strong("foo".to_owned()))
),
Ok(Some(true))
);
}
#[test]
fn test_none_match_none_none() {
assert_eq!(
none_match(&TestRequest::default().to_http_request(), None),
Ok(None)
);
}
#[test]
fn test_none_match_any_some() {
assert_eq!(
none_match(
&TestRequest::default()
.insert_header(("If-None-Match", "*"))
.to_http_request(),
Some(EntityTag::new_strong("foo".to_owned()))
),
Ok(Some(false))
);
}
#[test]
fn test_none_match_any_none() {
assert_eq!(
none_match(
&TestRequest::default()
.insert_header(("If-None-Match", "*"))
.to_http_request(),
None
),
Ok(Some(true))
);
}
#[test]
fn test_none_match_some_some() {
assert_eq!(
none_match(
&TestRequest::default()
.insert_header(("If-None-Match", "\"foo\""))
.to_http_request(),
Some(EntityTag::new_strong("foo".to_owned()))
),
Ok(Some(false))
);
}
#[test]
fn test_none_match_some_none() {
assert_eq!(
none_match(
&TestRequest::default()
.insert_header(("If-None-Match", "\"foo\""))
.to_http_request(),
None
),
Ok(Some(true))
);
}
#[test]
fn test_none_match_some_unknown() {
assert_eq!(
none_match(
&TestRequest::default()
.insert_header(("If-None-Match", "\"foo\""))
.to_http_request(),
Some(EntityTag::new_strong("bar".to_owned()))
),
Ok(Some(true))
);
}
}
server/src/handlers.rs
+135 -41
View File
@@ -6,7 +6,7 @@ use actix_web::{
dev::ServiceRequest,
http::{
self,
header::{self, ContentLength, ContentType},
header::{self, ContentLength, ContentType, EntityTag},
},
web::{Data, Header, Path, Payload},
};
@@ -17,14 +17,22 @@ use size::Size;
use tracing::*;
use uuid::Uuid;
use crate::conditional;
use crate::s3::S3Client;
use crate::{blob, merge};
use crate::{
blob,
conditional::{ConditionalMatch, any_match, none_match},
merge,
postgres::ObjectPart,
};
use crate::{
config::CONFIG,
postgres::{self, Pool},
};
use crate::{merge::MergeStrategy, recovery};
const CACHE_CONTROL: &str = "public, max-age=0, must-revalidate";
#[derive(Deserialize, Debug)]
pub struct ObjectPath {
workspace: Uuid,
@@ -45,6 +53,12 @@ pub enum ApiError {
#[error(transparent)]
ActixParseError(#[from] actix_web::error::ParseError),
#[error(transparent)]
ConditionalError(#[from] conditional::ConditionalError),
#[error("Precondition Failed")]
PreconditionFailed,
#[error(transparent)]
Other(#[from] anyhow::Error),
}
@@ -56,6 +70,12 @@ impl actix_web::error::ResponseError for ApiError {
match self {
ApiError::ActixError(error) => error.error_response(),
ApiError::ConditionalError(_) => HttpResponse::BadRequest().body("Bad Request"),
ApiError::PreconditionFailed => {
HttpResponse::PreconditionFailed().body("Precondition Failed")
}
_ => {
tracing::error!(error=?self, "Internal error in http handler");
HttpResponse::InternalServerError().body("Internal Server Error")
@@ -187,6 +207,10 @@ pub async fn put(request: HttpRequest, payload: Payload) -> HandlerResult<HttpRe
let pool = request.app_data::<Data<Pool>>().unwrap().to_owned();
let s3 = request.app_data::<Data<S3Client>>().unwrap().to_owned();
let parts = postgres::find_parts::<PartData>(&pool, path.workspace, &path.key).await?;
let conditionals = validate_put_conditionals(request.request(), &parts)?;
let uploaded = blob::upload(&s3, &pool, headers.content_length, payload).await?;
merge::validate_put_body(merge_strategy, &uploaded)?;
@@ -213,7 +237,8 @@ pub async fn put(request: HttpRequest, payload: Payload) -> HandlerResult<HttpRe
});
let obj_parts = vec![&part_data];
recovery::set_object(&s3, path.workspace, &part_data.key, obj_parts).await?;
recovery::set_object(&s3, path.workspace, &part_data.key, obj_parts, conditionals).await?;
postgres::set_part(&pool, path.workspace, &part_data.key, inline, &part_data).await?;
@@ -253,8 +278,9 @@ pub async fn patch(request: HttpRequest, payload: Payload) -> HandlerResult<Http
let parts = postgres::find_parts::<PartData>(&pool, path.workspace, &path.key).await?;
let mut response = if !parts.is_empty() {
let first = parts.first().unwrap();
let merge_strategy = first.data.merge_strategy.unwrap();
let conditionals = validate_patch_conditionals(request.request(), &parts)?;
let merge_strategy = objectpart_strategy(&parts).unwrap();
merge::validate_patch_request(merge_strategy, &headers)?;
@@ -288,7 +314,8 @@ pub async fn patch(request: HttpRequest, payload: Payload) -> HandlerResult<Http
.map(|p| &p.data)
.chain(std::iter::once(&part_data))
.collect::<Vec<&PartData>>();
recovery::set_object(&s3, path.workspace, &part_data.key, obj_parts).await?;
recovery::set_object(&s3, path.workspace, &part_data.key, obj_parts, conditionals).await?;
postgres::append_part(
&pool,
@@ -330,29 +357,38 @@ pub async fn get(request: HttpRequest) -> HandlerResult<HttpResponse> {
span.record("huly_key", &path.key);
let pool = request.app_data::<Data<Pool>>().unwrap().to_owned();
let s3 = request
.app_data::<Data<S3Client>>()
.unwrap()
.to_owned()
.into_inner();
let parts = postgres::find_parts::<PartData>(&pool, path.workspace, &path.key).await?;
let response = if !parts.is_empty() {
let mut response = HttpResponse::Ok();
let etag = objectpart_etag(&parts).unwrap();
let headers = parts[0].data.headers.as_ref();
if let Some(headers) = headers {
for (header, value) in headers.iter() {
response.insert_header((header.as_str(), value.to_owned()));
match none_match(request.request(), Some(etag.clone()))? {
Some(false) => HttpResponse::NotModified()
.insert_header((header::ETAG, etag))
.insert_header((header::CACHE_CONTROL, CACHE_CONTROL))
.finish(),
_ => {
let mut response = HttpResponse::Ok();
let s3 = request
.app_data::<Data<S3Client>>()
.unwrap()
.to_owned()
.into_inner();
let headers = parts[0].data.headers.as_ref();
if let Some(headers) = headers {
for (header, value) in headers.iter() {
response.insert_header((header.as_str(), value.to_owned()));
}
}
response.insert_header((header::ETAG, etag));
response.insert_header((header::CACHE_CONTROL, CACHE_CONTROL));
response.body(merge::stream(s3, parts).await?)
}
}
let etag = parts.last().unwrap().data.etag.to_owned();
response.insert_header((header::ETAG, etag));
response.body(merge::stream(s3, parts).await?)
} else {
HttpResponse::NotFound().finish()
};
@@ -376,28 +412,37 @@ pub async fn head(request: HttpRequest) -> HandlerResult<HttpResponse> {
let parts = postgres::find_parts::<PartData>(&pool, path.workspace, &path.key).await?;
let response = if !parts.is_empty() {
let mut response = HttpResponse::Ok();
let etag = objectpart_etag(&parts).unwrap();
let headers = parts[0].data.headers.as_ref();
if let Some(headers) = headers {
for (header, value) in headers.iter() {
response.insert_header((header.as_str(), value.to_owned()));
match none_match(request.request(), Some(etag.clone()))? {
Some(false) => HttpResponse::NotModified()
.insert_header((header::ETAG, etag))
.insert_header((header::CACHE_CONTROL, CACHE_CONTROL))
.finish(),
_ => {
let mut response = HttpResponse::Ok();
let headers = parts[0].data.headers.as_ref();
if let Some(headers) = headers {
for (header, value) in headers.iter() {
response.insert_header((header.as_str(), value.to_owned()));
}
}
response.insert_header((header::ETAG, etag));
response.insert_header((header::CACHE_CONTROL, CACHE_CONTROL));
// see https://github.com/actix/examples/blob/master/forms/multipart-s3/src/main.rs#L67-L79
let content_length = merge::content_length(parts);
match content_length {
Some(content_length) => response.body(SizedStream::new(
content_length as u64,
stream::empty::<Result<_, io::Error>>().boxed_local(),
)),
None => response.finish(),
}
}
}
let etag = parts.last().unwrap().data.etag.to_owned();
response.insert_header((header::ETAG, etag));
// see https://github.com/actix/examples/blob/master/forms/multipart-s3/src/main.rs#L67-L79
let content_length = merge::content_length(parts);
match content_length {
Some(content_length) => response.body(SizedStream::new(
content_length as u64,
stream::empty::<Result<_, io::Error>>().boxed_local(),
)),
None => response.finish(),
}
} else {
HttpResponse::NotFound().finish()
};
@@ -408,3 +453,52 @@ pub async fn head(request: HttpRequest) -> HandlerResult<HttpResponse> {
pub async fn delete(_path: Path<ObjectPath>) -> HandlerResult<HttpResponse> {
unimplemented!("delete is not implemented")
}
fn objectpart_etag(parts: &Vec<ObjectPart<PartData>>) -> Option<EntityTag> {
parts
.last()
.map(|p| EntityTag::new_strong(p.data.etag.to_owned()))
}
fn objectpart_strategy(parts: &Vec<ObjectPart<PartData>>) -> Option<MergeStrategy> {
parts.first().map(|p| p.data.merge_strategy.unwrap())
}
fn validate_patch_conditionals(
req: &HttpRequest,
parts: &Vec<ObjectPart<PartData>>,
) -> Result<Option<ConditionalMatch>, ApiError> {
let etag = objectpart_etag(parts);
match any_match(req, etag)? {
Some(false) => Err(ApiError::PreconditionFailed),
_ => {
let parts_data = parts.iter().map(|p| &p.data).collect::<Vec<&PartData>>();
let parts_etag = recovery::object_etag(parts_data)?;
Ok(Some(ConditionalMatch::IfMatch(parts_etag)))
}
}
}
fn validate_put_conditionals(
req: &HttpRequest,
parts: &Vec<ObjectPart<PartData>>,
) -> Result<Option<ConditionalMatch>, ApiError> {
let etag = objectpart_etag(parts);
match any_match(req, etag.clone())? {
Some(true) => {
let parts_data = parts.iter().map(|p| &p.data).collect::<Vec<&PartData>>();
let parts_etag = recovery::object_etag(parts_data)?;
Ok(Some(ConditionalMatch::IfMatch(parts_etag)))
}
Some(false) => Err(ApiError::PreconditionFailed),
None => match none_match(req, etag.clone())? {
Some(true) => Ok(Some(ConditionalMatch::IfNoneMatch("*".to_owned()))),
Some(false) => Err(ApiError::PreconditionFailed),
None => Ok(None),
},
}
}
server/src/main.rs
+2 -1
View File
@@ -15,12 +15,13 @@ use uuid::Uuid;
use hulyrs::services::jwt::actix::ServiceRequestExt;
mod blob;
mod conditional;
mod config;
mod handlers;
mod merge;
mod recovery;
mod patch;
mod postgres;
mod recovery;
mod s3;
use config::CONFIG;
server/src/recovery.rs
+19 -4
View File
@@ -1,26 +1,41 @@
use bytes::Bytes;
use crate::conditional::ConditionalMatch;
use crate::config::CONFIG;
use crate::{handlers::PartData, s3::S3Client};
pub fn object_etag(parts: Vec<&PartData>) -> anyhow::Result<String> {
let body = Bytes::from(serde_json::to_string(&parts)?);
let digest = md5::compute(body);
Ok(format!("{:x}", digest))
}
pub async fn set_object(
s3: &S3Client,
workspace: uuid::Uuid,
key: &str,
parts: Vec<&PartData>,
conditions: Option<ConditionalMatch>,
) -> anyhow::Result<()> {
let s3_bucket = &CONFIG.s3_bucket;
let key = format!("blob/{}/{}", workspace, key);
let body = Bytes::from(serde_json::to_string(&parts)?);
s3.put_object()
let mut cmd = s3
.put_object()
.bucket(s3_bucket)
.key(key)
.body(body.into())
.content_type("application/json")
.send()
.await?;
.content_type("application/json");
cmd = match conditions {
Some(ConditionalMatch::IfMatch(etag)) => cmd.if_match(etag),
Some(ConditionalMatch::IfNoneMatch(etag)) => cmd.if_none_match(etag),
None => cmd,
};
cmd.send().await?;
Ok(())
}