wip create w2l inference job

src/http_server/endpoints/w2l/enqueue_infer_w2l_with_uploads.rs

@@ -0,0 +1,347 @@
+//use bytes::{Bytes, BytesMut};
+use actix_http::http::header;
+use actix_multipart::{Multipart, Field};
+use actix_web::body::ResponseBody::Body;
+use actix_web::dev::HttpResponseBuilder;
+use actix_web::http::StatusCode;
+use actix_web::http::header::ContentDisposition;
+use actix_web::web::{Data, Json, BytesMut};
+use actix_web::{middleware, web, App, Error, HttpResponse, HttpServer, HttpRequest, Either, ResponseError};
+use anyhow::anyhow;
+use crate::buckets::bucket_client::BucketClient;
+use crate::buckets::bucket_paths::hash_to_bucket_path;
+use crate::http_server::web_utils::ip_address::get_request_ip;
+use crate::http_server::web_utils::read_multipart_field_bytes::checked_read_multipart_bytes;
+use crate::http_server::web_utils::read_multipart_field_bytes::read_multipart_field_as_boolean;
+use crate::http_server::web_utils::read_multipart_field_bytes::read_multipart_field_as_text;
+use crate::http_server::web_utils::read_multipart_field_bytes::read_multipart_field_bytes;
+use crate::server_state::ServerState;
+use crate::util::anyhow_result::AnyhowResult;
+use crate::util::random_prefix_crockford_token::random_prefix_crockford_token;
+use crate::util::random_token::random_token;
+use crate::util::random_uuid::generate_random_uuid;
+use derive_more::{Display, Error};
+use futures::{StreamExt, TryStreamExt};
+use log::{warn, info};
+use sqlx::MySqlPool;
+use sqlx::error::Error::Database;
+use std::io::Write;
+use std::sync::Arc;
+use uuid::Uuid;
+
+const BUCKET_AUDIO_FILE_NAME : &'static str = "input_audio_file";
+const BUCKET_IMAGE_FILE_NAME: &'static str = "input_image_file";
+const BUCKET_VIDEO_FILE_NAME : &'static str = "input_video_file";
+
+const MIN_BYTES : usize = 10;
+const MAX_BYTES : usize = 1024 * 1024 * 20;
+
+
+
+/// Just to query for existence
+#[derive(Serialize)]
+pub struct W2lTemplateExistenceRecord {
+  pub template_token: String,
+}
+
+#[derive(Serialize)]
+pub struct InferW2lWithUploadSuccessResponse {
+  pub success: bool,
+  pub job_token: String,
+}
+
+#[derive(Serialize)]
+pub struct ErrorResponse {
+  pub success: bool,
+  pub error_reason: String,
+}
+
+#[derive(Debug, Display)]
+pub enum InferW2lWithUploadError {
+  BadInput(String),
+  EmptyFileUploaded,
+  ServerError,
+}
+
+impl ResponseError for InferW2lWithUploadError {
+  fn status_code(&self) -> StatusCode {
+    match *self {
+      InferW2lWithUploadError::BadInput(_) => StatusCode::BAD_REQUEST,
+      InferW2lWithUploadError::EmptyFileUploaded => StatusCode::BAD_REQUEST,
+      InferW2lWithUploadError::ServerError => StatusCode::INTERNAL_SERVER_ERROR,
+    }
+  }
+
+  fn error_response(&self) -> HttpResponse {
+    let error_reason = match self {
+      InferW2lWithUploadError::BadInput(reason) => reason.to_string(),
+      InferW2lWithUploadError::EmptyFileUploaded => "empty file uploaded".to_string(),
+      InferW2lWithUploadError::ServerError => "server error".to_string(),
+    };
+
+    let response = ErrorResponse {
+      success: false,
+      error_reason,
+    };
+
+    let body = match serde_json::to_string(&response) {
+      Ok(json) => json,
+      Err(_) => "{}".to_string(),
+    };
+
+    HttpResponseBuilder::new(self.status_code())
+      .set_header(header::CONTENT_TYPE, "application/json")
+      .body(body)
+  }
+}
+
+/// This handles audio uploads w/ W2L templates.
+pub async fn enqueue_infer_w2l_with_uploads(
+  http_request: HttpRequest,
+  server_state: web::Data<Arc<ServerState>>,
+  mut payload: Multipart)
+  //-> Result<Json<UploadResponse>, HttpResponse>
+  -> Result<HttpResponse, InferW2lWithUploadError>
+{
+  // ==================== TODO: READ IP AND RATE LIMIT ==================== //
+
+  let ip_address = get_request_ip(&http_request);
+
+  // TODO: Check IP address holds
+  // TODO: Rate Limiting!
+  /*let mut hold_exists = handler_state.redis_client.get_ip_hold(&ip_address)
+    .map_err(|err| HttpResponse::InternalServerError()
+      .body(format!("Error: {:?}", err)))?;
+
+  match ip_address.as_ref() {
+    "127.0.0.1" => {
+      warn!("Allowing IP address hold bypass for {}", &ip_address);
+      hold_exists = false
+    },
+    _ => {},
+  }
+
+  if hold_exists {
+    return Err(HttpResponse::TooManyRequests().body("you already have pending requests").into())
+  }*/
+
+  // ==================== READ MULTIPART REQUEST ==================== //
+
+  let mut maybe_uuid_idempotency_token: Option<String> = None;
+  let mut maybe_template_token: Option<String> = None;
+  let mut maybe_audio_file_name : Option<String> = None;
+
+  while let Ok(Some(mut field)) = payload.try_next().await {
+    let content_disposition: ContentDisposition = field.content_disposition().unwrap();
+    let field_name = content_disposition.get_name().unwrap_or("");
+    let filename = content_disposition.get_filename().unwrap_or("");
+
+    match field_name {
+      "uuid_idempotency_token" => {
+        // Form text field.
+        maybe_uuid_idempotency_token = read_multipart_field_as_text(&mut field).await
+          .map_err(|_| InferW2lWithUploadError::ServerError)?;
+      },
+      "template_token" => {
+        // Form text field.
+        maybe_template_token = read_multipart_field_as_text(&mut field).await
+          .map_err(|_| InferW2lWithUploadError::ServerError)?;
+      },
+      "audio" => {
+        // Form binary data.
+        maybe_audio_file_name = Some(filename.to_string());
+
+        let maybe_bytes = checked_read_multipart_bytes(&mut field).await
+          .map_err(|_| InferW2lWithUploadError::ServerError)?;
+
+        let bytes = match maybe_bytes {
+          Some(bytes) => bytes,
+          None => {
+            return Err(InferW2lWithUploadError::EmptyFileUploaded); // Nothing was uploaded!
+          },
+        };
+      },
+      _ => continue,
+    }
+
+    info!("Saved file: {}.", &filename);
+  }
+
+  // ==================== CHECK REQUEST ==================== //
+
+  let template_token = match &maybe_template_token {
+    Some(ref token) => token.to_string(),
+    None => {
+      return Err(InferW2lWithUploadError::BadInput("No template selected".to_string()));
+    }
+  };
+
+  let uuid_idempotency_token = match maybe_uuid_idempotency_token {
+    Some(token) => token,
+    None => {
+      return Err(InferW2lWithUploadError::BadInput("No uuid idempotency token".to_string()));
+    }
+  };
+
+  let exists = check_template_exists(&template_token, &server_state.mysql_pool).await
+    .map_err(|e| {
+      warn!("error checking tmpl existence : {:?}", e);
+      InferW2lWithUploadError::ServerError
+    })?;
+
+  if !exists {
+    return Err(InferW2lWithUploadError::BadInput("Template does not exist".to_string()));
+  }
+
+  // ==================== READ SESSION ==================== //
+
+  let maybe_user_session = server_state
+    .session_checker
+    .maybe_get_user_session(&http_request, &server_state.mysql_pool)
+    .await
+    .map_err(|e| {
+      warn!("Session checker error: {:?}", e);
+      InferW2lWithUploadError::ServerError
+    })?;
+
+  let mut maybe_user_token : Option<String> = maybe_user_session
+    .as_ref()
+    .map(|user_session| user_session.user_token.to_string());
+
+
+  // ==================== TODO: UPLOAD AUDIO FILE ==================== //
+
+  let upload_uuid = generate_random_uuid();
+
+  let audio_upload_bucket_path = hash_to_bucket_path(
+    &upload_uuid,
+    Some("/audio_uploads") // TODO: pull from configs
+  ).map_err(|e| {
+    warn!("Hash bucket path error: {:?}", e);
+    InferW2lWithUploadError::ServerError
+  })?;
+
+  // TODO: Upload audio to bucket!
+
+  // ==================== SAVE JOB RECORD ==================== //
+
+  // This token is returned to the client.
+  let job_token = random_prefix_crockford_token("TIJ:", 32)
+    .map_err(|e| {
+      warn!("Error creating token");
+      InferW2lWithUploadError::ServerError
+    })?;
+
+  info!("Creating w2l inference job record...");
+
+  let query_result = sqlx::query!(
+        r#"
+INSERT INTO w2l_inference_jobs
+SET
+  token = ?,
+  uuid_idempotency_token = ?,
+
+  maybe_w2l_template_token = ?,
+  maybe_audio_bucket_location = ?,
+
+  maybe_creator_user_token = ?,
+  creator_ip_address = ?,
+  disable_end_bump = false,
+  creator_set_visibility = "public",
+  status = "pending"
+        "#,
+        job_token.to_string(),
+        uuid_idempotency_token.to_string(),
+        maybe_template_token.clone(),
+        Some(audio_upload_bucket_path.clone()), // TODO THIS ISN'T WIDE ENOUGH
+        maybe_user_token.clone(),
+        ip_address.to_string()
+    )
+    .execute(&server_state.mysql_pool)
+    .await;
+
+  let record_id = match query_result {
+    Ok(res) => {
+      res.last_insert_id()
+    },
+    Err(err) => {
+      warn!("New w2l template upload creation DB error: {:?}", err);
+
+      // NB: SQLSTATE[23000]: Integrity constraint violation
+      // NB: MySQL Error Code 1062: Duplicate key insertion (this is harder to access)
+      match err {
+        Database(err) => {
+          let maybe_code = err.code().map(|c| c.into_owned());
+          /*match maybe_code.as_deref() {
+            Some("23000") => {
+              if err.message().contains("username") {
+                return Err(UsernameTaken);
+              } else if err.message().contains("email_address") {
+                return Err(EmailTaken);
+              }
+            }
+            _ => {},
+          }*/
+        },
+        _ => {},
+      }
+      return Err(InferW2lWithUploadError::ServerError);
+    }
+  };
+
+  // TODO: IP Address holds.
+  //handler_state.redis_client.set_ip_hold(&ip_address)
+  //  .map_err(|err| HttpResponse::InternalServerError()
+  //    .body(format!("Redis Err: {:?}", err)))?;
+
+  let response = InferW2lWithUploadSuccessResponse {
+    success: true,
+    job_token: job_token.to_string(),
+  };
+
+  let body = serde_json::to_string(&response)
+    .map_err(|e| InferW2lWithUploadError::ServerError)?;
+
+  Ok(HttpResponse::Ok()
+    .content_type("application/json")
+    .body(body))
+}
+
+async fn check_template_exists(template_token: &str, mysql_pool: &MySqlPool) -> AnyhowResult<bool>
+{
+  // NB: Lookup failure is Err(RowNotFound).
+  // NB: Since this is publicly exposed, we don't query sensitive data.
+  let maybe_template = sqlx::query_as!(
+      W2lTemplateExistenceRecord,
+        r#"
+SELECT
+    token as template_token
+FROM w2l_templates
+WHERE token = ?
+AND deleted_at IS NULL
+        "#,
+      &template_token
+    )
+    .fetch_one(mysql_pool)
+    .await; // TODO: This will return error if it doesn't exist
+
+  let record_exists = match maybe_template {
+    Ok(record) => {
+      true
+    },
+    Err(err) => {
+      match err {
+        RowNotFound => {
+          false
+        },
+        _ => {
+          warn!("Infer w2l query error: {:?}", err);
+          return Err(anyhow!("infer w2l query error: {:?}", err));
+        }
+      }
+    }
+  };
+
+  Ok(record_exists)
+}
+

src/http_server/endpoints/w2l/mod.rs

@@ -1,4 +1,5 @@
 pub mod enqueue_infer_w2l;
+pub mod enqueue_infer_w2l_with_uploads;
 pub mod enqueue_upload_w2l_template;
 pub mod get_w2l_template;
 pub mod list_w2l_templates;

src/http_server/web_utils/mod.rs

@@ -1,3 +1,4 @@
 pub mod cookie_manager;
 pub mod ip_address;
+pub mod read_multipart_field_bytes;
 pub mod session_checker;

src/http_server/web_utils/read_multipart_field_bytes.rs

@@ -0,0 +1,66 @@
+//use bytes::BytesMut;
+use actix_multipart::Field;
+use actix_web::web::BytesMut;
+use anyhow::anyhow;
+use futures::{StreamExt, TryStreamExt};
+use log::warn;
+
+const MIN_BYTES : usize = 10;
+const MAX_BYTES : usize = 1024 * 1024 * 20;
+
+/// Read any field from a multipart form as bytes, but do some error checking.
+pub async fn checked_read_multipart_bytes(field: &mut Field) -> anyhow::Result<Option<BytesMut>> {
+  let bytes = read_multipart_field_bytes(field).await?;
+
+  if bytes.is_empty() {
+    // This field is empty
+    return Ok(None);
+  }
+
+  if bytes.len() < MIN_BYTES {
+    warn!("Bytes of upload too few: {}", bytes.len());
+    return Err(anyhow!("Too few bytes"));
+  }
+
+  if bytes.len() > MAX_BYTES {
+    warn!("Bytes of upload too much: {}", bytes.len());
+    return Err(anyhow!("Too many bytes"));
+  }
+
+  Ok(Some(bytes))
+}
+
+/// Read any field from a multipart form as bytes.
+pub async fn read_multipart_field_bytes(field: &mut Field) -> anyhow::Result<BytesMut> {
+  let mut bytes = BytesMut::new();
+  while let Some(chunk) = field.next().await {
+    let chunk_data = chunk.map_err(|err| anyhow!("Error: {:?}", err))?;
+    bytes.extend_from_slice(&chunk_data);
+  }
+  Ok(bytes)
+}
+
+/// This reads inputs, text boxes, etc. because form-multipart is hard.
+pub async fn read_multipart_field_as_text(field: &mut Field) -> anyhow::Result<Option<String>> {
+  let bytes = read_multipart_field_bytes(field).await?;
+  let value = String::from_utf8(bytes.to_vec())?;
+
+  if value.is_empty() || value == "none" {
+    return Ok(None);
+  }
+
+  Ok(Some(value))
+}
+
+/// Read a html form boolean (eg. input type=checkbox); assumes default is false.
+pub async fn read_multipart_field_as_boolean(field: &mut Field) -> anyhow::Result<bool> {
+  let result = read_multipart_field_as_text(field).await?;
+  Ok(match result.as_ref().map(String::as_str) {
+    Some("true") => true,
+    Some("TRUE") => true,
+    Some("false") => false,
+    Some("FALSE") => false,
+    None => false,
+    _ => true,
+  })
+}