migrations, about to insert media file result

_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;

_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;

_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;

_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;

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);
+  }
+}

crates/_types/buckets/src/public/media_files/mod.rs

@@ -0,0 +1,2 @@
+pub mod directory;
+pub mod original_file;

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);
+  }
+}

crates/_types/buckets/src/public/mod.rs

@@ -1,3 +1,4 @@
 pub mod media_uploads;
 pub mod public_path;
-pub mod voice_conversion_results;
+pub mod voice_conversion_results;
+pub mod media_files;

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<Self, String> {
     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);

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<Self, String> {
+    match value {
+      "media_file" => Ok(Self::MediaFile),
+      "lipsync_animation" => Ok(Self::LipsyncAnimation),
+      _ => Err(format!("invalid value: {:?}", value)),
+    }
+  }
+
+  pub fn all_variants() -> BTreeSet<Self> {
+    // 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());
+  }
+}

crates/_types/enums/src/by_table/generic_synthetic_ids/mod.rs

@@ -0,0 +1 @@
+pub mod id_category;

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;

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,

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<u64> = 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))
+}

crates/lib/mysql_queries/src/queries/media_files/mod.rs

@@ -0,0 +1 @@
+pub mod insert_media_file;

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;

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<JobSuccess
   let file_size_bytes = file_size(&output_video_fs_path)
       .map_err(|err| ProcessSingleJobError::Other(err))?;
 
-  let maybe_mimetype = get_mimetype_for_file(&output_video_fs_path)
+  let mimetype = get_mimetype_for_file(&output_video_fs_path)
       .map_err(|err| ProcessSingleJobError::from_io_error(err))?
-      .map(|mime| mime.to_string());
+      .map(|mime| mime.to_string())
+      .ok_or(ProcessSingleJobError::Other(anyhow!("Mimetype could not be determined")))?;
 
 
-  info!("TEST FILE...");
-
-  thread::sleep(Duration::from_secs(300));
-
-  /*let audio_info = decode_basic_audio_file_info(&output_video_fs_path, maybe_mimetype.as_deref(), None)
-      .map_err(|err| ProcessSingleJobError::Other(err))?;
-
-  if audio_info.required_full_decode {
-    warn!("Required a full decode of the output data to get duration! That's inefficient!");
-  }
-
-  // TODO: Make a new python image that generates spectrograms from any audio file.
-
-  let file_metadata = FileMetadata {
-    duration_millis: audio_info.duration_millis,
-    mimetype: maybe_mimetype,
-    file_size_bytes,
-  };
-
-  //safe_delete_temp_file(&output_metadata_fs_path);
-
   // ==================== UPLOAD AUDIO TO BUCKET ==================== //
 
   job_progress_reporter.log_status("uploading result")
       .map_err(|e| ProcessSingleJobError::Other(e))?;
 
-  let result_bucket_location = VoiceConversionResultOriginalFilePath::generate_new();
+  let result_bucket_location = MediaFileBucketPath::generate_new();
 
   let result_bucket_object_pathbuf = result_bucket_location.to_full_object_pathbuf();
 
   info!("Audio destination bucket path: {:?}", &result_bucket_object_pathbuf);
 
-  info!("Uploading audio...");
+  info!("Uploading media ...");
 
   args.job_dependencies.public_bucket_client.upload_filename_with_content_type(
     &result_bucket_object_pathbuf,
     &output_video_fs_path,
-    "audio/wav")
+    &mimetype) // TODO: We should check the mimetype to make sure bad payloads can't get uploaded
       .await
       .map_err(|e| ProcessSingleJobError::Other(e))?;
 
+  // ==================== DELETE TEMP FILES ==================== //
+
   safe_delete_temp_file(&output_video_fs_path);
 
-//  // ==================== UPLOAD SPECTROGRAM TO BUCKETS ==================== //
-//
-//  let spectrogram_result_object_path = args.job_dependencies.bucket_path_unifier.tts_inference_spectrogram_output_path(
-//    &job.uuid_idempotency_token); // TODO: Don't use this!
-//
-//  info!("Spectrogram destination bucket path: {:?}", &spectrogram_result_object_path);
-//
-//  info!("Uploading spectrogram...");
-//
-//  args.job_dependencies.public_bucket_client.upload_filename_with_content_type(
-//    &spectrogram_result_object_path,
-//    &output_spectrogram_fs_path,
-//    "application/json")
-//      .await
-//      .map_err(|e| ProcessSingleJobError::Other(e))?;
-//
-//  safe_delete_temp_file(&output_spectrogram_fs_path);
-
-  // ==================== DELETE DOWNLOADED FILE ==================== //
-
   // NB: We should be using a tempdir, but to make absolutely certain we don't overflow the disk...
   safe_delete_temp_directory(&work_temp_dir);
 
   // ==================== SAVE RECORDS ==================== //
 
-  info!("Saving vc inference record...");
+  info!("Saving SadTalker result (media_files table record) ...");
 
+  /* TODO
   let (inference_result_token, id) = insert_voice_conversion_result(InsertArgs {
     pool: &args.job_dependencies.mysql_pool,
     job: &job,
     public_bucket_hash: result_bucket_location.get_object_hash(),
-    file_size_bytes: file_metadata.file_size_bytes,
-    duration_millis: file_metadata.duration_millis.unwrap_or(0),
+    file_size_bytes,
+    duration_millis: 0,
     is_on_prem: args.job_dependencies.container.is_on_prem,
     worker_hostname: &args.job_dependencies.container.hostname,
     worker_cluster: &args.job_dependencies.container.cluster_name,
@@ -255,14 +219,16 @@ pub async fn process_job(args: SadTalkerProcessJobArgs<'_>) -> Result<JobSuccess
   })
       .await
       .map_err(|e| ProcessSingleJobError::Other(e))?;
+      
+   */
 
-  info!("VC Done.");
+  info!("SadTalker Done.");
 
   // TODO: Update upstream to be strongly typed
   let maybe_user_token = job.maybe_creator_user_token.as_deref()
       .map(|token| UserToken::new_from_str(token));
 
-  args.job_dependencies.firehose_publisher.vc_inference_finished(
+  args.job_dependencies.firehose_publisher.lipsync_animation_finished(
     maybe_user_token.as_ref(),
     &job.inference_job_token,
     inference_result_token.as_str())
@@ -280,30 +246,9 @@ pub async fn process_job(args: SadTalkerProcessJobArgs<'_>) -> Result<JobSuccess
 
   Ok(JobSuccessResult {
     maybe_result_entity: Some(ResultEntity {
-      entity_type: InferenceResultType::VoiceConversion,
+      entity_type: InferenceResultType::MediaFile,
       entity_token: inference_result_token.to_string(),
     }),
     inference_duration,
   })
-
-   */
-
-
-  // TODO - REMOVE -
-  let inference_start_time = Instant::now();
-  let inference_duration = Instant::now().duration_since(inference_start_time);
-  Ok(JobSuccessResult {
-    maybe_result_entity: Some(ResultEntity {
-      entity_type: InferenceResultType::VoiceConversion,
-      entity_token: "todo".to_string(),
-    }),
-    inference_duration,
-  })
-}
-
-#[derive(Deserialize, Default)]
-struct FileMetadata {
-  pub duration_millis: Option<u64>,
-  pub mimetype: Option<String>,
-  pub file_size_bytes: u64,
 }