upsert media file upload endpoint

This commit is contained in:
Brandon Thomas committed 2024-03-26 03:11:15 -04:00
1 parent 8aa64a9ae6
commit 4f471cf7b7
11 files changed
+633 -9

No files matched your search

+1
View File
@@ -44,6 +44,7 @@
<w>transcodings</w>
<w>unpermitted</w>
<w>upserting</w>
<w>upserts</w>
<w>vits</w>
<w>vocoded</w>
<w>vocoder</w>
@@ -6,3 +6,4 @@ pub mod get_media_file_for_inference;
pub mod list;
pub mod rename_media_file;
pub mod update_media_file;
pub mod upsert;
@@ -0,0 +1 @@
pub mod upsert_media_file_from_file_upload;
@@ -0,0 +1,233 @@
use anyhow::anyhow;
use sqlx;
use sqlx::MySqlPool;
use enums::by_table::generic_synthetic_ids::id_category::IdCategory;
use enums::by_table::media_files::media_file_class::MediaFileClass;
use enums::by_table::media_files::media_file_origin_category::MediaFileOriginCategory;
use enums::by_table::media_files::media_file_origin_product_category::MediaFileOriginProductCategory;
use enums::by_table::media_files::media_file_subtype::MediaFileSubtype;
use enums::by_table::media_files::media_file_type::MediaFileType;
use enums::common::visibility::Visibility;
use errors::AnyhowResult;
use tokens::tokens::anonymous_visitor_tracking::AnonymousVisitorTrackingToken;
use tokens::tokens::media_files::MediaFileToken;
use tokens::tokens::users::UserToken;
use crate::queries::generic_synthetic_ids::transactional_increment_generic_synthetic_id::transactional_increment_generic_synthetic_id;
pub enum UploadType {
Filesystem,
DeviceCaptureApi,
StorytellerEngine,
}
pub struct UpsertMediaFileFromUploadArgs<'a> {
pub pool: &'a MySqlPool,
/// If the media file token is present, we're upserting an existing media file.
pub maybe_media_file_token: Option<&'a MediaFileToken>,
pub maybe_creator_user_token: Option<&'a UserToken>,
pub maybe_creator_anonymous_visitor_token: Option<&'a AnonymousVisitorTrackingToken>,
pub creator_ip_address: &'a str,
pub creator_set_visibility: Visibility,
pub upload_type: UploadType,
pub media_file_type: MediaFileType,
pub maybe_media_class: Option<MediaFileClass>,
pub maybe_media_subtype: Option<MediaFileSubtype>,
pub maybe_mime_type: Option<&'a str>,
pub file_size_bytes: u64,
pub duration_millis: u64,
pub sha256_checksum: &'a str,
// TODO: Media duration.
//pub duration_millis: u64,
pub public_bucket_directory_hash: &'a str,
pub maybe_public_bucket_prefix: Option<&'a str>,
pub maybe_public_bucket_extension: Option<&'a str>,
}
pub async fn upsert_media_file_from_file_upload(
args: UpsertMediaFileFromUploadArgs<'_>
) -> AnyhowResult<(MediaFileToken, u64)>
{
let token = args.maybe_media_file_token
.map(|token| token.clone())
.unwrap_or_else(|| MediaFileToken::generate());
let origin_category = match args.upload_type {
UploadType::Filesystem => MediaFileOriginCategory::Upload,
UploadType::DeviceCaptureApi => MediaFileOriginCategory::DeviceApi,
UploadType::StorytellerEngine => MediaFileOriginCategory::StoryEngine,
};
let mut maybe_creator_file_synthetic_id : Option<u64> = None;
let mut maybe_creator_category_synthetic_id : Option<u64> = None;
let mut transaction = args.pool.begin().await?;
if let Some(user_token) = args.maybe_creator_user_token.as_deref() {
let next_media_file_id = transactional_increment_generic_synthetic_id(
&user_token,
IdCategory::MediaFile,
&mut transaction
).await?;
let next_voice_conversion_id = transactional_increment_generic_synthetic_id(
&user_token,
IdCategory::FileUpload,
&mut transaction
).await?;
maybe_creator_file_synthetic_id = Some(next_media_file_id);
maybe_creator_category_synthetic_id = Some(next_voice_conversion_id);
}
const ORIGIN_PRODUCT_CATEGORY : MediaFileOriginProductCategory = MediaFileOriginProductCategory::Unknown;
let media_class = args.maybe_media_class.unwrap_or(MediaFileClass::Unknown);
let record_id = {
let query_result = sqlx::query!(
r#"
INSERT INTO media_files
SET
token = ?,
origin_category = ?,
origin_product_category = ?,
maybe_origin_model_type = NULL,
maybe_origin_model_token = NULL,
media_type = ?,
media_class = ?,
maybe_media_subtype = ?,
maybe_mime_type = ?,
file_size_bytes = ?,
checksum_sha2 = ?,
public_bucket_directory_hash = ?,
maybe_public_bucket_prefix = ?,
maybe_public_bucket_extension = ?,
maybe_creator_user_token = ?,
maybe_creator_anonymous_visitor_token = ?,
creator_ip_address = ?,
creator_set_visibility = ?,
maybe_creator_file_synthetic_id = ?,
maybe_creator_category_synthetic_id = ?
ON DUPLICATE KEY UPDATE
origin_category = ?,
origin_product_category = ?,
maybe_origin_model_type = NULL,
maybe_origin_model_token = NULL,
media_type = ?,
media_class = ?,
maybe_media_subtype = ?,
maybe_mime_type = ?,
file_size_bytes = ?,
checksum_sha2 = ?,
public_bucket_directory_hash = ?,
maybe_public_bucket_prefix = ?,
maybe_public_bucket_extension = ?,
maybe_creator_user_token = ?,
maybe_creator_anonymous_visitor_token = ?,
creator_ip_address = ?,
creator_set_visibility = ?,
maybe_creator_file_synthetic_id = ?,
maybe_creator_category_synthetic_id = ?
"#,
token.as_str(),
origin_category.to_str(),
ORIGIN_PRODUCT_CATEGORY.to_str(),
args.media_file_type.to_str(),
media_class.to_str(),
args.maybe_media_subtype.map(|s| s.to_str()),
args.maybe_mime_type,
args.file_size_bytes,
args.sha256_checksum,
args.public_bucket_directory_hash,
args.maybe_public_bucket_prefix,
args.maybe_public_bucket_extension,
args.maybe_creator_user_token,
args.maybe_creator_anonymous_visitor_token,
args.creator_ip_address,
args.creator_set_visibility.to_str(),
maybe_creator_file_synthetic_id,
maybe_creator_category_synthetic_id,
// Duplicate key
origin_category.to_str(),
ORIGIN_PRODUCT_CATEGORY.to_str(),
args.media_file_type.to_str(),
media_class.to_str(),
args.maybe_media_subtype.map(|s| s.to_str()),
args.maybe_mime_type,
args.file_size_bytes,
args.sha256_checksum,
args.public_bucket_directory_hash,
args.maybe_public_bucket_prefix,
args.maybe_public_bucket_extension,
args.maybe_creator_user_token,
args.maybe_creator_anonymous_visitor_token,
args.creator_ip_address,
args.creator_set_visibility.to_str(),
maybe_creator_file_synthetic_id,
maybe_creator_category_synthetic_id,
)
.execute(&mut *transaction)
.await;
let record_id = match query_result {
Ok(res) => {
res.last_insert_id()
},
Err(err) => {
// TODO: handle better
//transaction.rollback().await?;
return Err(anyhow!("Mysql error: {:?}", err));
}
};
record_id
};
transaction.commit().await?;
Ok((token, record_id))
}
@@ -5,4 +5,5 @@ pub mod get_media_file;
pub mod list;
pub mod rename_media_file_handler;
pub mod update_media_file;
pub mod upload;
pub mod upload;
pub mod upsert_write;
@@ -0,0 +1,2 @@
pub mod write_engine_asset;
pub mod write_error;
@@ -0,0 +1,106 @@
use actix_multipart::Multipart;
use actix_web::web::BytesMut;
use anyhow::anyhow;
use futures::TryStreamExt;
use log::warn;
use enums::by_table::media_files::media_file_class::MediaFileClass;
use enums::by_table::media_files::media_file_subtype::MediaFileSubtype;
use errors::AnyhowResult;
use tokens::tokens::media_files::MediaFileToken;
use tokens::tokens::zs_voice_datasets::ZsVoiceDatasetToken;
use crate::http_server::web_utils::read_multipart_field_bytes::{checked_read_multipart_bytes, read_multipart_field_as_text};
pub struct MediaFileUploadData {
pub uuid_idempotency_token: Option<String>,
pub file_name: Option<String>,
pub file_bytes: Option<BytesMut>,
pub media_file_token: Option<MediaFileToken>,
pub media_file_subtype: Option<MediaFileSubtype>,
pub media_file_class: Option<MediaFileClass>,
}
/// Pull common parts out of multipart media HTTP requests, typically for handling file uploads.
pub async fn drain_multipart_request(mut multipart_payload: Multipart) -> AnyhowResult<MediaFileUploadData> {
let mut uuid_idempotency_token = None;
let mut file_bytes = None;
let mut file_name = None;
let mut media_file_token = None;
let mut media_file_class = None;
let mut media_file_subtype = None;
while let Ok(Some(mut field)) = multipart_payload.try_next().await {
let mut field_name = None;
let mut field_filename = None;
if let content_disposition = field.content_disposition() {
field_name = content_disposition.get_name()
.map(|s| s.to_string());
field_filename = content_disposition.get_filename() // NB: Only used for the file bytes.
.map(|s| s.to_string());
}
match field_name.as_deref() {
Some("uuid_idempotency_token") => {
uuid_idempotency_token = read_multipart_field_as_text(&mut field).await
.map_err(|err| {
warn!("Error reading idempotency token: {:?}", &err);
err
})?;
},
Some("file") => {
file_name = field_filename.clone();
file_bytes = checked_read_multipart_bytes(&mut field).await
.map_err(|err| {
warn!("Error reading audio upload: {:?}", &err);
err
})?;
},
Some("media_file_token") => {
media_file_token = read_multipart_field_as_text(&mut field).await
.map_err(|err| {
warn!("Error reading source: {:?}", &err);
err
})?
.map(|field| MediaFileToken::new_from_str(&field));
},
Some("media_file_subtype") => {
media_file_subtype = read_multipart_field_as_text(&mut field).await
.map_err(|err| {
warn!("Error reading source: {:?}", &err);
err
})?
.map(|field| MediaFileSubtype::from_str(&field))
.transpose()
.map_err(|err| {
warn!("Wrong MediaFileSubtype: {:?}", &err);
anyhow!("Wrong MediaFileSubtype: {:?}", &err)
})?;
},
Some("media_file_class") => {
media_file_class = read_multipart_field_as_text(&mut field).await
.map_err(|err| {
warn!("Error reading source: {:?}", &err);
err
})?
.map(|field| MediaFileClass::from_str(&field))
.transpose()
.map_err(|err| {
warn!("Wrong MediaFileClass: {:?}", &err);
anyhow!("Wrong MediaFileClass: {:?}", &err)
})?;
},
_ => continue,
}
}
Ok(MediaFileUploadData {
uuid_idempotency_token,
file_name,
file_bytes,
media_file_token,
media_file_class,
media_file_subtype,
})
}
@@ -0,0 +1,2 @@
pub mod drain_multipart_request;
pub mod write_engine_asset_media_file_handler;
@@ -0,0 +1,239 @@
use std::collections::HashSet;
use std::io::{BufReader, Cursor};
use std::path::PathBuf;
use std::sync::Arc;
use actix_multipart::Multipart;
use actix_web::{HttpRequest, HttpResponse, web};
use log::{error, info, warn};
use once_cell::sync::Lazy;
use utoipa::ToSchema;
use buckets::public::media_files::bucket_file_path::MediaFileBucketPath;
use enums::by_table::media_files::media_file_type::MediaFileType;
use enums::common::visibility::Visibility;
use hashing::sha256::sha256_hash_bytes::sha256_hash_bytes;
use http_server_common::request::get_request_ip::get_request_ip;
use mimetypes::mimetype_for_bytes::get_mimetype_for_bytes;
use mimetypes::mimetype_to_extension::mimetype_to_extension;
use mysql_queries::queries::idepotency_tokens::insert_idempotency_token::insert_idempotency_token;
use mysql_queries::queries::media_files::upsert::upsert_media_file_from_file_upload::{upsert_media_file_from_file_upload, UpsertMediaFileFromUploadArgs, UploadType};
use tokens::tokens::media_files::MediaFileToken;
use videos::get_mp4_info::{get_mp4_info, get_mp4_info_for_bytes, get_mp4_info_for_bytes_and_len};
use crate::http_server::endpoints::media_files::upsert_write::write_engine_asset::drain_multipart_request::drain_multipart_request;
use crate::http_server::endpoints::media_files::upsert_write::write_error::MediaFileWriteError;
use crate::server_state::ServerState;
use crate::validations::validate_idempotency_token_format::validate_idempotency_token_format;
// Unlike the "upload" endpoints, which are pure inserts, these endpoints are *upserts*.
#[derive(Serialize, ToSchema)]
pub struct WriteEngineAssetMediaSuccessResponse {
pub success: bool,
pub media_file_token: MediaFileToken,
}
#[utoipa::path(
post,
path = "/v1/media_files/write/engine_asset",
responses(
(status = 200, description = "Success Update", body = WriteEngineAssetMediaSuccessResponse),
(status = 400, description = "Bad input", body = MediaFileWriteError),
(status = 401, description = "Not authorized", body = MediaFileWriteError),
(status = 429, description = "Too many requests", body = MediaFileWriteError),
(status = 500, description = "Server error", body = MediaFileWriteError),
),
params(
("request" = (), description = "Ask Brandon. This is form-multipart."),
)
)]
pub async fn write_engine_asset_media_file_handler(
http_request: HttpRequest,
server_state: web::Data<Arc<ServerState>>,
mut multipart_payload: Multipart,
) -> Result<HttpResponse, MediaFileWriteError> {
let mut mysql_connection = server_state.mysql_pool
.acquire()
.await
.map_err(|err| {
error!("MySql pool error: {:?}", err);
MediaFileWriteError::ServerError
})?;
// ==================== READ SESSION ==================== //
let maybe_user_session = server_state
.session_checker
.maybe_get_user_session_from_connection(&http_request, &mut mysql_connection)
.await
.map_err(|e| {
error!("Session checker error: {:?}", e);
MediaFileWriteError::ServerError
})?;
let maybe_avt_token = server_state
.avt_cookie_manager
.get_avt_token_from_request(&http_request);
// ==================== BANNED USERS ==================== //
if let Some(ref user) = maybe_user_session {
if user.is_banned {
return Err(MediaFileWriteError::NotAuthorized);
}
}
// ==================== RATE LIMIT ==================== //
let rate_limiter = match maybe_user_session {
None => &server_state.redis_rate_limiters.file_upload_logged_out,
Some(ref _session) => &server_state.redis_rate_limiters.file_upload_logged_in,
};
if let Err(_err) = rate_limiter.rate_limit_request(&http_request) {
return Err(MediaFileWriteError::RateLimited);
}
// ==================== READ MULTIPART REQUEST ==================== //
let upload_media_request = drain_multipart_request(multipart_payload)
.await
.map_err(|e| {
// TODO: Error handling could be nicer.
MediaFileWriteError::BadInput("bad request".to_string())
})?;
// TODO(bt, 2024-02-26): This should be a transaction.
let uuid_idempotency_token = upload_media_request.uuid_idempotency_token
.ok_or(MediaFileWriteError::BadInput("no uuid".to_string()))?;
// ==================== HANDLE IDEMPOTENCY ==================== //
if let Err(reason) = validate_idempotency_token_format(&uuid_idempotency_token) {
return Err(MediaFileWriteError::BadInput(reason));
}
insert_idempotency_token(&uuid_idempotency_token, &mut *mysql_connection)
.await
.map_err(|err| {
error!("Error inserting idempotency token: {:?}", err);
MediaFileWriteError::BadInput("invalid idempotency token".to_string())
})?;
// ==================== UPLOAD METADATA ==================== //
let creator_set_visibility = maybe_user_session
.as_ref()
.map(|user_session| user_session.preferred_tts_result_visibility) // TODO: We need a new type of visibility control.
.unwrap_or(Visibility::default());
// ==================== USER DATA ==================== //
let ip_address = get_request_ip(&http_request);
let maybe_user_token = maybe_user_session
.map(|session| session.get_strongly_typed_user_token());
// ==================== FILE DATA ==================== //
let file_bytes = match upload_media_request.file_bytes {
None => return Err(MediaFileWriteError::BadInput("missing file contents".to_string())),
Some(bytes) => bytes,
};
let maybe_filename = upload_media_request.file_name
.as_deref()
.map(|filename| PathBuf::from(filename));
let maybe_file_extension = maybe_filename
.as_ref()
.and_then(|filename| filename.extension())
.and_then(|ext| ext.to_str());
let (suffix, media_file_type, mimetype) = match maybe_file_extension {
None => {
return Err(MediaFileWriteError::BadInput("no file extension".to_string()));
}
Some("bvh") => (".bvh", MediaFileType::Bvh, "application/octet-stream"),
Some("fbx") => (".fbx", MediaFileType::Fbx, "application/octet-stream"),
Some("glb") => (".glb", MediaFileType::Glb, "application/octet-stream"),
Some("gltf") => (".gltf", MediaFileType::Gltf, "application/octet-stream"),
Some("ron") => (".scn.ron", MediaFileType::SceneRon, "application/octet-stream"),
_ => {
return Err(MediaFileWriteError::BadInput(
"unsupported file extension. Must be bvh, glb, gltf, or fbx.".to_string()));
}
};
let file_size_bytes = file_bytes.len();
let hash = sha256_hash_bytes(&file_bytes)
.map_err(|io_error| {
error!("Problem hashing bytes: {:?}", io_error);
MediaFileWriteError::ServerError
})?;
// ==================== UPLOAD AND SAVE ==================== //
// TODO(bt,2024-03-26): At first I thought we should map these to the existing file paths on upsert,
// but now I'm thinking we can just lead cruft in the bucket and clean it later. We don't have the
// benefit of restoring old versions (if we mapped to existing paths but had a versioning scheme),
// but we can move fast.
const PREFIX : Option<&str> = Some("upload_");
let public_upload_path = MediaFileBucketPath::generate_new(PREFIX, Some(suffix));
info!("Uploading media to bucket path: {}", public_upload_path.get_full_object_path_str());
server_state.public_bucket_client.upload_file_with_content_type(
public_upload_path.get_full_object_path_str(),
file_bytes.as_ref(),
mimetype)
.await
.map_err(|e| {
warn!("Upload media bytes to bucket error: {:?}", e);
MediaFileWriteError::ServerError
})?;
// TODO(bt, 2024-02-22): This should be a transaction.
let (token, record_id) = upsert_media_file_from_file_upload(UpsertMediaFileFromUploadArgs {
maybe_media_file_token: upload_media_request.media_file_token.as_ref(),
maybe_creator_user_token: maybe_user_token.as_ref(),
maybe_creator_anonymous_visitor_token: maybe_avt_token.as_ref(),
creator_ip_address: &ip_address,
creator_set_visibility,
upload_type: UploadType::Filesystem,
media_file_type,
maybe_media_class: upload_media_request.media_file_class,
maybe_media_subtype: upload_media_request.media_file_subtype,
maybe_mime_type: Some(mimetype),
file_size_bytes: file_size_bytes as u64,
duration_millis: 0,
sha256_checksum: &hash,
public_bucket_directory_hash: public_upload_path.get_object_hash(),
maybe_public_bucket_prefix: PREFIX,
maybe_public_bucket_extension: Some(suffix),
pool: &server_state.mysql_pool,
})
.await
.map_err(|err| {
warn!("New file creation DB error: {:?}", err);
MediaFileWriteError::ServerError
})?;
info!("new media file id: {} token: {:?}", record_id, &token);
let response = WriteEngineAssetMediaSuccessResponse {
success: true,
media_file_token: token,
};
let body = serde_json::to_string(&response)
.map_err(|e| MediaFileWriteError::ServerError)?;
return Ok(HttpResponse::Ok()
.content_type("application/json")
.body(body));
}
@@ -0,0 +1,37 @@
use actix_web::error::ResponseError;
use actix_web::http::StatusCode;
use actix_web::HttpResponse;
use utoipa::ToSchema;
use http_server_common::response::serialize_as_json_error::serialize_as_json_error;
#[derive(Debug, Serialize, ToSchema)]
pub enum MediaFileWriteError {
BadInput(String),
NotAuthorized,
MustBeLoggedIn,
ServerError,
RateLimited,
}
impl ResponseError for MediaFileWriteError {
fn status_code(&self) -> StatusCode {
match *self {
MediaFileWriteError::BadInput(_) => StatusCode::BAD_REQUEST,
MediaFileWriteError::NotAuthorized => StatusCode::UNAUTHORIZED,
MediaFileWriteError::MustBeLoggedIn => StatusCode::UNAUTHORIZED,
MediaFileWriteError::ServerError => StatusCode::INTERNAL_SERVER_ERROR,
MediaFileWriteError::RateLimited => StatusCode::TOO_MANY_REQUESTS,
}
}
fn error_response(&self) -> HttpResponse {
serialize_as_json_error(self)
}
}
impl std::fmt::Display for MediaFileWriteError {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
write!(f, "{:?}", self)
}
}
@@ -16,6 +16,7 @@ use crate::http_server::endpoints::media_files::update_media_file::update_media_
use crate::http_server::endpoints::media_files::upload::upload_generic::upload_media_file_handler::upload_media_file_handler;
use crate::http_server::endpoints::media_files::upload::upload_engine_asset::upload_engine_asset_media_file_handler::upload_engine_asset_media_file_handler;
use crate::http_server::endpoints::media_files::upload::upload_video::upload_video_media_file_handler::upload_video_media_file_handler;
use crate::http_server::endpoints::media_files::upsert_write::write_engine_asset::write_engine_asset_media_file_handler::write_engine_asset_media_file_handler;
pub fn add_media_file_routes<T, B> (app: App<T>) -> App<T>
where
@@ -75,13 +76,13 @@ pub fn add_media_file_routes<T, B> (app: App<T>) -> App<T>
.route(web::post().to(upload_engine_asset_media_file_handler))
.route(web::head().to(|| HttpResponse::Ok()))
)
// .service(web::resource("/write/engine_asset")
// .route(web::post().to(/* todo */))
// .route(web::head().to(|| HttpResponse::Ok()))
// )
// .service(web::resource("/write/scene_file")
// .route(web::post().to(/* todo */))
// .route(web::head().to(|| HttpResponse::Ok()))
// )
.service(web::resource("/write/engine_asset")
.route(web::post().to(write_engine_asset_media_file_handler))
.route(web::head().to(|| HttpResponse::Ok()))
)
//.service(web::resource("/write/scene_file")
// .route(web::post().to(/* todo */))
// .route(web::head().to(|| HttpResponse::Ok()))
//)
)
}