From 359b7a5bd093ff8c024e9c1153768eb0cea8ea9c Mon Sep 17 00:00:00 2001 From: Brandon Thomas Date: Fri, 15 Sep 2023 02:19:41 -0400 Subject: [PATCH] migrations, about to insert media file result --- .../down.sql | 5 + .../up.sql | 31 ++++ .../down.sql | 9 + .../up.sql | 11 ++ .../src/public/media_files/directory.rs | 75 ++++++++ .../buckets/src/public/media_files/mod.rs | 2 + .../src/public/media_files/original_file.rs | 101 +++++++++++ crates/_types/buckets/src/public/mod.rs | 3 +- .../inference_result_type.rs | 14 +- .../generic_synthetic_ids/id_category.rs | 93 ++++++++++ .../src/by_table/generic_synthetic_ids/mod.rs | 1 + crates/_types/enums/src/by_table/mod.rs | 1 + .../src/mediators/firehose_publisher.rs | 11 ++ .../queries/media_files/insert_media_file.rs | 167 ++++++++++++++++++ .../src/queries/media_files/mod.rs | 1 + crates/lib/mysql_queries/src/queries/mod.rs | 1 + .../lipsync/sad_talker/process_job.rs | 91 ++-------- 17 files changed, 542 insertions(+), 75 deletions(-) create mode 100644 _sql/migrations/2023-09-15-053054_generic_synthetic_ids/down.sql create mode 100644 _sql/migrations/2023-09-15-053054_generic_synthetic_ids/up.sql create mode 100644 _sql/migrations/2023-09-15-055106_alter_media_files_synthetic_ids/down.sql create mode 100644 _sql/migrations/2023-09-15-055106_alter_media_files_synthetic_ids/up.sql create mode 100644 crates/_types/buckets/src/public/media_files/directory.rs create mode 100644 crates/_types/buckets/src/public/media_files/mod.rs create mode 100644 crates/_types/buckets/src/public/media_files/original_file.rs create mode 100644 crates/_types/enums/src/by_table/generic_synthetic_ids/id_category.rs create mode 100644 crates/_types/enums/src/by_table/generic_synthetic_ids/mod.rs create mode 100644 crates/lib/mysql_queries/src/queries/media_files/insert_media_file.rs create mode 100644 crates/lib/mysql_queries/src/queries/media_files/mod.rs diff --git a/_sql/migrations/2023-09-15-053054_generic_synthetic_ids/down.sql b/_sql/migrations/2023-09-15-053054_generic_synthetic_ids/down.sql new file mode 100644 index 0000000000..d5b0e4596c --- /dev/null +++ b/_sql/migrations/2023-09-15-053054_generic_synthetic_ids/down.sql @@ -0,0 +1,5 @@ +-- noinspection SqlDialectInspectionForFile +-- noinspection SqlNoDataSourceInspectionForFile +-- noinspection SqlResolveForFile + +drop table generic_synthetic_ids; diff --git a/_sql/migrations/2023-09-15-053054_generic_synthetic_ids/up.sql b/_sql/migrations/2023-09-15-053054_generic_synthetic_ids/up.sql new file mode 100644 index 0000000000..5554ad9d5b --- /dev/null +++ b/_sql/migrations/2023-09-15-053054_generic_synthetic_ids/up.sql @@ -0,0 +1,31 @@ +-- noinspection SqlDialectInspectionForFile +-- noinspection SqlNoDataSourceInspectionForFile +-- noinspection SqlResolveForFile + +-- It gets unwieldy to show results without a numeric sequence. +-- We create a synthetic sequence for each user to make revisiting results easier. +-- This should be a reusable table for handling this for many different types of results. +CREATE TABLE generic_synthetic_ids ( + -- Not used for anything except replication. + id BIGINT(20) NOT NULL AUTO_INCREMENT, + + -- Effective "primary key" (PUBLIC) + user_token VARCHAR(32) NOT NULL, + + -- The type of synthetic id to increment. + -- eg "media_file", "lipsync_animation", etc. + id_category VARCHAR(32) NOT NULL, + + -- The next result id. + -- After we use an id, we increment this counter. + next_id BIGINT(20) NOT NULL DEFAULT 0, + + created_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP, + updated_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP, + + -- INDICES -- + PRIMARY KEY (id), + UNIQUE KEY (user_token, id_category), + KEY fk_user_token (user_token) + +) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_bin; diff --git a/_sql/migrations/2023-09-15-055106_alter_media_files_synthetic_ids/down.sql b/_sql/migrations/2023-09-15-055106_alter_media_files_synthetic_ids/down.sql new file mode 100644 index 0000000000..93a24238d3 --- /dev/null +++ b/_sql/migrations/2023-09-15-055106_alter_media_files_synthetic_ids/down.sql @@ -0,0 +1,9 @@ +-- noinspection SqlDialectInspectionForFile +-- noinspection SqlNoDataSourceInspectionForFile +-- noinspection SqlResolveForFile + +-- Revert the rename +ALTER TABLE media_files RENAME COLUMN maybe_creator_file_synthetic_id TO maybe_creator_synthetic_id ; + +ALTER TABLE media_files +DROP COLUMN maybe_creator_category_synthetic_id; diff --git a/_sql/migrations/2023-09-15-055106_alter_media_files_synthetic_ids/up.sql b/_sql/migrations/2023-09-15-055106_alter_media_files_synthetic_ids/up.sql new file mode 100644 index 0000000000..e5bcb82988 --- /dev/null +++ b/_sql/migrations/2023-09-15-055106_alter_media_files_synthetic_ids/up.sql @@ -0,0 +1,11 @@ +-- noinspection SqlDialectInspectionForFile +-- noinspection SqlNoDataSourceInspectionForFile +-- noinspection SqlResolveForFile + +-- We'll have one ordering by files +ALTER TABLE media_files RENAME COLUMN maybe_creator_synthetic_id TO maybe_creator_file_synthetic_id; + +-- And another ordering by category +ALTER TABLE media_files + ADD COLUMN maybe_creator_category_synthetic_id BIGINT(20) DEFAULT NULL + AFTER maybe_creator_file_synthetic_id; diff --git a/crates/_types/buckets/src/public/media_files/directory.rs b/crates/_types/buckets/src/public/media_files/directory.rs new file mode 100644 index 0000000000..b0e628cb80 --- /dev/null +++ b/crates/_types/buckets/src/public/media_files/directory.rs @@ -0,0 +1,75 @@ +use crate::public::public_path::PublicPath; +use crate::util::hashed_directory_path_long_string::hashed_directory_path_long_string; +use crockford::crockford_entropy_lower; + +// TODO: Generate these from a macro. + +// TODO: Use a central path registry for quick reference +const MEDIA_FILE_DIRECTORY: &str = "/media"; + +/// Directory for user media uploads. +/// Each uploaded file gets its own directory so that we can store the original +/// file alongside re-processed derivatives. +#[derive(Clone)] +pub struct MediaFileBucketDirectory { + object_hash: String, + directory: String, +} + +impl PublicPath for MediaFileBucketDirectory {} + +impl MediaFileBucketDirectory { + + pub fn generate_new() -> Self { + let entropy = crockford_entropy_lower(32); + Self::from_object_hash(&entropy) + } + + pub fn from_object_hash(object_hash: &str) -> Self { + // TODO: Path construction could be cleaner. + let middle = hashed_directory_path_long_string(object_hash); + let directory = format!("{}/{}{}", MEDIA_FILE_DIRECTORY, middle, object_hash); + Self { + object_hash: object_hash.to_string(), + directory, + } + } + + pub fn get_directory_path_str(&self) -> &str { + &self.directory + } + + pub fn get_object_hash(&self) -> &str { + &self.object_hash + } +} + +#[cfg(test)] +mod tests { + use crate::public::media_files::directory::MediaFileBucketDirectory; + + #[test] + pub fn generate_new_entropy() { + let directory = MediaFileBucketDirectory::generate_new(); + assert_eq!(directory.get_object_hash().len(), 32); + } + + #[test] + pub fn get_directory_path_str() { + let directory = MediaFileBucketDirectory::from_object_hash("abcdefghijk"); + assert_eq!(directory.get_directory_path_str(), "/media/a/b/c/d/e/abcdefghijk"); + } + + #[test] + pub fn get_directory_path_str_short_name() { + let directory = MediaFileBucketDirectory::from_object_hash("foo"); + assert_eq!(directory.get_directory_path_str(), "/media/f/o/foo"); + } + + #[test] + pub fn get_object_hash() { + let hash = "abcdefghijk"; + let directory = MediaFileBucketDirectory::from_object_hash(hash); + assert_eq!(directory.get_object_hash(), hash); + } +} diff --git a/crates/_types/buckets/src/public/media_files/mod.rs b/crates/_types/buckets/src/public/media_files/mod.rs new file mode 100644 index 0000000000..42ede393cf --- /dev/null +++ b/crates/_types/buckets/src/public/media_files/mod.rs @@ -0,0 +1,2 @@ +pub mod directory; +pub mod original_file; \ No newline at end of file diff --git a/crates/_types/buckets/src/public/media_files/original_file.rs b/crates/_types/buckets/src/public/media_files/original_file.rs new file mode 100644 index 0000000000..1a65bc07c0 --- /dev/null +++ b/crates/_types/buckets/src/public/media_files/original_file.rs @@ -0,0 +1,101 @@ +use crate::public::media_files::directory::MediaFileBucketDirectory; +use crate::public::public_path::PublicPath; +use crockford::crockford_entropy_lower; +use std::path::PathBuf; + +const ORIGINAL_FILE_BASENAME : &str = "file.bin"; + +// TODO: Generate these from a macro. + +/// The original user upload file. +/// It may have derivative files (down samples, crops, etc.) that live alongside it. +#[derive(Clone)] +pub struct MediaFileBucketPath { + directory: MediaFileBucketDirectory, + full_object_path: String, +} + +impl PublicPath for MediaFileBucketPath {} + +impl MediaFileBucketPath { + + pub fn generate_new() -> Self { + let entropy = crockford_entropy_lower(32); + Self::from_object_hash(&entropy) + } + + pub fn from_object_hash(hash: &str) -> Self { + // TODO: Path construction could be cleaner. + let directory = MediaFileBucketDirectory::from_object_hash(hash); + let full_object_path = format!("{}/{}", directory.get_directory_path_str(), ORIGINAL_FILE_BASENAME); + Self { + directory, + full_object_path, + } + } + + pub fn get_full_object_path_str(&self) -> &str { + &self.full_object_path + } + + pub fn to_full_object_pathbuf(&self) -> PathBuf { + PathBuf::from(&self.full_object_path) + } + + pub fn get_directory(&self) -> &MediaFileBucketDirectory { + &self.directory + } + + pub fn get_object_hash(&self) -> &str { + &self.directory.get_object_hash() + } + + pub fn get_basename() -> &'static str { + ORIGINAL_FILE_BASENAME + } +} + +#[cfg(test)] +mod tests { + use crate::public::media_files::original_file::MediaFileBucketPath; + use std::path::PathBuf; + + #[test] + pub fn generate_new_entropy() { + let file = MediaFileBucketPath::generate_new(); + assert_eq!(file.get_object_hash().len(), 32); + assert_eq!(file.get_directory().get_object_hash().len(), 32); + } + + #[test] + pub fn get_full_object_path_str() { + let file = MediaFileBucketPath::from_object_hash("abcdefghijk"); + assert_eq!(file.get_full_object_path_str(), "/media/a/b/c/d/e/abcdefghijk/file.bin"); + } + + #[test] + pub fn to_full_object_pathbuf() { + let file = MediaFileBucketPath::from_object_hash("abcdefghijk"); + assert_eq!(file.to_full_object_pathbuf(), PathBuf::from("/media/a/b/c/d/e/abcdefghijk/file.bin")); + } + + #[test] + pub fn get_full_object_path_str_short_name() { + let file = MediaFileBucketPath::from_object_hash("foo"); + assert_eq!(file.get_full_object_path_str(), "/media/f/o/foo/file.bin"); + } + + #[test] + pub fn get_full_object_path_str_starts_with_directory() { + let file = MediaFileBucketPath::from_object_hash("abcdefghijk"); + let full_path = file.get_full_object_path_str(); + assert!(full_path.starts_with(file.get_directory().get_directory_path_str())) + } + + #[test] + pub fn get_object_hash() { + let hash = "abcdefghijk"; + let file = MediaFileBucketPath::from_object_hash(hash); + assert_eq!(file.get_object_hash(), hash); + } +} diff --git a/crates/_types/buckets/src/public/mod.rs b/crates/_types/buckets/src/public/mod.rs index f27eafae3d..c0879da257 100644 --- a/crates/_types/buckets/src/public/mod.rs +++ b/crates/_types/buckets/src/public/mod.rs @@ -1,3 +1,4 @@ pub mod media_uploads; pub mod public_path; -pub mod voice_conversion_results; \ No newline at end of file +pub mod voice_conversion_results; +pub mod media_files; \ No newline at end of file diff --git a/crates/_types/enums/src/by_table/generic_inference_jobs/inference_result_type.rs b/crates/_types/enums/src/by_table/generic_inference_jobs/inference_result_type.rs index 6f2a9244e8..ba254aa315 100644 --- a/crates/_types/enums/src/by_table/generic_inference_jobs/inference_result_type.rs +++ b/crates/_types/enums/src/by_table/generic_inference_jobs/inference_result_type.rs @@ -17,6 +17,11 @@ use strum::EnumIter; #[cfg_attr(test, derive(EnumIter, EnumCount))] #[derive(Clone, Copy, Eq, PartialEq, Hash, Ord, PartialOrd, Deserialize, Serialize)] pub enum InferenceResultType { + /// Result is stored in the "media_files" table. + /// (The upstream model could have produced any type of media - image, video, audio. That is irrelevant.) + #[serde(rename = "media_file")] + MediaFile, + #[serde(rename = "text_to_speech")] TextToSpeech, @@ -32,6 +37,7 @@ impl_mysql_enum_coders!(InferenceResultType); impl InferenceResultType { pub fn to_str(&self) -> &'static str { match self { + Self::MediaFile => "media_file", Self::TextToSpeech => "text_to_speech", Self::VoiceConversion => "voice_conversion", } @@ -39,6 +45,7 @@ impl InferenceResultType { pub fn from_str(value: &str) -> Result { match value { + "media_file" => Ok(Self::MediaFile), "text_to_speech" => Ok(Self::TextToSpeech), "voice_conversion" => Ok(Self::VoiceConversion), _ => Err(format!("invalid value: {:?}", value)), @@ -49,6 +56,7 @@ impl InferenceResultType { // NB: BTreeSet is sorted // NB: BTreeSet::from() isn't const, but not worth using LazyStatic, etc. BTreeSet::from([ + InferenceResultType::MediaFile, InferenceResultType::TextToSpeech, InferenceResultType::VoiceConversion, ]) @@ -62,18 +70,21 @@ mod tests { #[test] fn test_serialization() { + assert_serialization(InferenceResultType::MediaFile, "media_file"); assert_serialization(InferenceResultType::TextToSpeech, "text_to_speech"); assert_serialization(InferenceResultType::VoiceConversion, "voice_conversion"); } #[test] fn to_str() { + assert_eq!(InferenceResultType::MediaFile.to_str(), "media_file"); assert_eq!(InferenceResultType::TextToSpeech.to_str(), "text_to_speech"); assert_eq!(InferenceResultType::VoiceConversion.to_str(), "voice_conversion"); } #[test] fn from_str() { + assert_eq!(InferenceResultType::from_str("media_file").unwrap(), InferenceResultType::MediaFile); assert_eq!(InferenceResultType::from_str("text_to_speech").unwrap(), InferenceResultType::TextToSpeech); assert_eq!(InferenceResultType::from_str("voice_conversion").unwrap(), InferenceResultType::VoiceConversion); } @@ -82,7 +93,8 @@ mod tests { fn all_variants() { // Static check let mut variants = InferenceResultType::all_variants(); - assert_eq!(variants.len(), 2); + assert_eq!(variants.len(), 3); + assert_eq!(variants.pop_first(), Some(InferenceResultType::MediaFile)); assert_eq!(variants.pop_first(), Some(InferenceResultType::TextToSpeech)); assert_eq!(variants.pop_first(), Some(InferenceResultType::VoiceConversion)); assert_eq!(variants.pop_first(), None); diff --git a/crates/_types/enums/src/by_table/generic_synthetic_ids/id_category.rs b/crates/_types/enums/src/by_table/generic_synthetic_ids/id_category.rs new file mode 100644 index 0000000000..dc9cdb2241 --- /dev/null +++ b/crates/_types/enums/src/by_table/generic_synthetic_ids/id_category.rs @@ -0,0 +1,93 @@ +use std::collections::BTreeSet; + +#[cfg(test)] +use strum::EnumCount; +#[cfg(test)] +use strum::EnumIter; + + +/// Used in the `generic_synthetic_ids` table in `VARCHAR(32)` field `id_category`. +/// +/// This lets us create synthetic increment IDs on a per-user, per-category basis. +/// +/// DO NOT CHANGE VALUES WITHOUT A MIGRATION STRATEGY. +#[cfg_attr(test, derive(EnumIter, EnumCount))] +#[derive(Clone, Copy, Eq, PartialEq, Hash, Ord, PartialOrd, Deserialize, Serialize)] +pub enum IdCategory { + /// media_files table + #[serde(rename = "media_file")] + MediaFile, + + /// Results from lipsync animations (which may live in the media_files table) + #[serde(rename = "lipsync_animation")] + LipsyncAnimation, +} + +// TODO(bt, 2022-12-21): This desperately needs MySQL integration tests! +impl_enum_display_and_debug_using_to_str!(IdCategory); +impl_mysql_enum_coders!(IdCategory); + +/// NB: Legacy API for older code. +impl IdCategory { + pub fn to_str(&self) -> &'static str { + match self { + Self::MediaFile => "media_file", + Self::LipsyncAnimation => "lipsync_animation", + } + } + + pub fn from_str(value: &str) -> Result { + match value { + "media_file" => Ok(Self::MediaFile), + "lipsync_animation" => Ok(Self::LipsyncAnimation), + _ => Err(format!("invalid value: {:?}", value)), + } + } + + pub fn all_variants() -> BTreeSet { + // NB: BTreeSet is sorted + // NB: BTreeSet::from() isn't const, but not worth using LazyStatic, etc. + BTreeSet::from([ + Self::MediaFile, + Self::LipsyncAnimation, + ]) + } +} + +#[cfg(test)] +mod tests { + use crate::by_table::generic_synthetic_ids::id_category::IdCategory; + use crate::test_helpers::assert_serialization; + + #[test] + fn test_serialization() { + assert_serialization(IdCategory::MediaFile, "media_file"); + assert_serialization(IdCategory::LipsyncAnimation, "lipsync_animation"); + } + + #[test] + fn to_str() { + assert_eq!(IdCategory::MediaFile.to_str(), "media_file"); + assert_eq!(IdCategory::LipsyncAnimation.to_str(), "lipsync_animation"); + } + + #[test] + fn from_str() { + assert_eq!(IdCategory::from_str("media_file").unwrap(), IdCategory::MediaFile); + assert_eq!(IdCategory::from_str("lipsync_animation").unwrap(), IdCategory::LipsyncAnimation); + } + + #[test] + fn all_variants() { + // Static check + let mut variants = IdCategory::all_variants(); + assert_eq!(variants.len(), 2); + assert_eq!(variants.pop_first(), Some(IdCategory::MediaFile)); + assert_eq!(variants.pop_first(), Some(IdCategory::LipsyncAnimation)); + assert_eq!(variants.pop_first(), None); + + // Generated check + use strum::IntoEnumIterator; + assert_eq!(IdCategory::all_variants().len(), IdCategory::iter().len()); + } +} diff --git a/crates/_types/enums/src/by_table/generic_synthetic_ids/mod.rs b/crates/_types/enums/src/by_table/generic_synthetic_ids/mod.rs new file mode 100644 index 0000000000..17a815f89c --- /dev/null +++ b/crates/_types/enums/src/by_table/generic_synthetic_ids/mod.rs @@ -0,0 +1 @@ +pub mod id_category; \ No newline at end of file diff --git a/crates/_types/enums/src/by_table/mod.rs b/crates/_types/enums/src/by_table/mod.rs index 0ff5757fd4..eb4d0997a9 100644 --- a/crates/_types/enums/src/by_table/mod.rs +++ b/crates/_types/enums/src/by_table/mod.rs @@ -3,6 +3,7 @@ pub mod audit_logs; pub mod comments; pub mod generic_download_jobs; pub mod generic_inference_jobs; +pub mod generic_synthetic_ids; pub mod media_files; pub mod media_uploads; pub mod model_categories; diff --git a/crates/lib/mysql_queries/src/mediators/firehose_publisher.rs b/crates/lib/mysql_queries/src/mediators/firehose_publisher.rs index a5e2346843..f39d3b86bb 100644 --- a/crates/lib/mysql_queries/src/mediators/firehose_publisher.rs +++ b/crates/lib/mysql_queries/src/mediators/firehose_publisher.rs @@ -236,6 +236,17 @@ impl FirehosePublisher { Ok(()) } + // TODO: Change result token type. + pub async fn lipsync_animation_finished(&self, maybe_user_token: Option<&UserToken>, inference_job_token: &InferenceJobToken, result_token: &str) -> AnyhowResult<()> { + let _record_id = self.insert( + FirehoseEvent::LipsyncAnimationCompleted, + maybe_user_token.map(|u| u.as_str()), + Some(inference_job_token.as_str()), // TODO: Is this pointing to the right entity? + Some(result_token) + ).await?; + Ok(()) + } + pub async fn enqueue_generic_download(&self, user_token: &str, job_token: &str) -> AnyhowResult<()> { let _record_id = self.insert( FirehoseEvent::GenericDownloadStarted, diff --git a/crates/lib/mysql_queries/src/queries/media_files/insert_media_file.rs b/crates/lib/mysql_queries/src/queries/media_files/insert_media_file.rs new file mode 100644 index 0000000000..cf413e28fa --- /dev/null +++ b/crates/lib/mysql_queries/src/queries/media_files/insert_media_file.rs @@ -0,0 +1,167 @@ +use anyhow::anyhow; +use crate::queries::generic_inference::job::list_available_generic_inference_jobs::AvailableInferenceJob; +use errors::AnyhowResult; +use log::warn; +use sqlx::MySqlPool; +use sqlx; +use enums::by_table::voice_conversion_results::voice_conversion_media_token_type::VoiceConversionMediaTokenType; +use tokens::tokens::voice_conversion_results::VoiceConversionResultToken; + +/// Used to give user-facing order to logged in user inference requests +pub struct SyntheticIdRecord { + pub next_id: i64, +} + +pub struct InsertArgs<'a> { + pub pool: &'a MySqlPool, + pub job: &'a AvailableInferenceJob, + pub public_bucket_hash: &'a str, + pub file_size_bytes: u64, + pub duration_millis: u64, + pub is_on_prem: bool, + pub worker_hostname: &'a str, + pub worker_cluster: &'a str, + pub is_debug_worker: bool, +} + +pub async fn insert_voice_conversion_result( + args: InsertArgs<'_> +) -> AnyhowResult<(VoiceConversionResultToken, u64)> +{ + let result_token = VoiceConversionResultToken::generate(); + + let mut maybe_creator_synthetic_id : Option = None; + + let mut transaction = args.pool.begin().await?; + + if let Some(creator_user_token) = args.job.maybe_creator_user_token.as_deref() { + let query_result = sqlx::query!( + r#" +INSERT INTO voice_conversion_result_synthetic_ids +SET + user_token = ?, + next_id = 1 +ON DUPLICATE KEY UPDATE + user_token = ?, + next_id = next_id + 1 + "#, + creator_user_token, + creator_user_token + ) + .execute(&mut transaction) + .await; + + match query_result { + Ok(_) => {}, + Err(err) => { + //transaction.rollback().await?; + warn!("Transaction failure: {:?}", err); + } + } + + let query_result = sqlx::query_as!( + SyntheticIdRecord, + r#" +SELECT + next_id +FROM + voice_conversion_result_synthetic_ids +WHERE + user_token = ? +LIMIT 1 + "#, + creator_user_token, + ) + .fetch_one(&mut transaction) + .await; + + let record : SyntheticIdRecord = match query_result { + Ok(record) => record, + Err(err) => { + warn!("Transaction failure: {:?}", err); + transaction.rollback().await?; + return Err(anyhow!("Transaction failure: {:?}", err)); + } + }; + + let next_id = record.next_id as u64; + maybe_creator_synthetic_id = Some(next_id); + } + + let vc_model_token = args.job.maybe_model_token.as_deref(); + let creator_ip_address = args.job.creator_ip_address.as_str(); + let creator_set_visibility = args.job.creator_set_visibility.clone(); + + let record_id = { + let query_result = sqlx::query!( + r#" +INSERT INTO voice_conversion_results +SET + token = ?, + + model_token = ?, + + media_token = ?, + media_token_type = ?, + + maybe_creator_user_token = ?, + maybe_creator_synthetic_id = ?, + + creator_ip_address = ?, + creator_set_visibility = ?, + + public_bucket_hash = ?, + bucket_has_wav = true, + + file_size_bytes = ?, + duration_millis = ?, + + is_generated_on_prem = ?, + generated_by_worker = ?, + generated_by_cluster = ?, + is_debug_request = ? + "#, + result_token.as_str(), + + args.job.maybe_model_token, + + args.job.maybe_input_source_token, + VoiceConversionMediaTokenType::MediaUpload.to_str(), //args.job.maybe_input_source_token_type, + + args.job.maybe_creator_user_token, + maybe_creator_synthetic_id, + + args.job.creator_ip_address, + args.job.creator_set_visibility.to_str(), + + args.public_bucket_hash, + + args.file_size_bytes, + args.duration_millis, + + args.is_on_prem, + args.worker_hostname, + args.worker_cluster, + args.is_debug_worker, + ) + .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((result_token, record_id)) +} diff --git a/crates/lib/mysql_queries/src/queries/media_files/mod.rs b/crates/lib/mysql_queries/src/queries/media_files/mod.rs new file mode 100644 index 0000000000..d63e0567e4 --- /dev/null +++ b/crates/lib/mysql_queries/src/queries/media_files/mod.rs @@ -0,0 +1 @@ +pub mod insert_media_file; \ No newline at end of file diff --git a/crates/lib/mysql_queries/src/queries/mod.rs b/crates/lib/mysql_queries/src/queries/mod.rs index 0a46b0904d..c4f762e10e 100644 --- a/crates/lib/mysql_queries/src/queries/mod.rs +++ b/crates/lib/mysql_queries/src/queries/mod.rs @@ -6,6 +6,7 @@ pub mod generic_download; pub mod generic_inference; pub mod health_check; pub mod ip_bans; +pub mod media_files; pub mod media_uploads; pub mod model_categories; pub mod public_event_feed; diff --git a/crates/service/job/inference_job/src/job/job_types/lipsync/sad_talker/process_job.rs b/crates/service/job/inference_job/src/job/job_types/lipsync/sad_talker/process_job.rs index d58f4eed06..88e9041e73 100644 --- a/crates/service/job/inference_job/src/job/job_types/lipsync/sad_talker/process_job.rs +++ b/crates/service/job/inference_job/src/job/job_types/lipsync/sad_talker/process_job.rs @@ -27,6 +27,7 @@ use mysql_queries::queries::voice_conversion::results::insert_voice_conversion_r use std::path::PathBuf; use std::thread; use std::time::{Duration, Instant}; +use buckets::public::media_files::original_file::MediaFileBucketPath; use tokens::users::user::UserToken; use crate::job::job_types::lipsync::sad_talker::sad_talker_inference_command::InferenceArgs; @@ -167,87 +168,50 @@ pub async fn process_job(args: SadTalkerProcessJobArgs<'_>) -> Result) -> Result) -> Result, - pub mimetype: Option, - pub file_size_bytes: u64, }