more firehose event types

src/common_queries/firehose_publisher.rs

@@ -59,44 +59,52 @@ pub struct FirehosePublisher {
 impl FirehosePublisher {
 
   pub async fn publish_user_sign_up(&self, user_token: &str) -> AnyhowResult<()> {
-    let token = random_prefix_crockford_token("EV", 32)?;
-    let event_type = FirehoseEvent::UserSignUp;
-
     let _record_id = self.insert(
-      event_type,
+    FirehoseEvent::UserSignUp,
       Some(user_token),
       Some(user_token)
     ).await?;
-
     Ok(())
   }
 
-  pub async fn publish_w2l_template_upload_enqueue(&self, user_token: &str, job_token: &str) -> AnyhowResult<()> {
-    let token = random_prefix_crockford_token("EV", 32)?;
-    let event_type = FirehoseEvent::W2lTemplateUploadStarted;
-
+  pub async fn enqueue_w2l_template_upload(&self, user_token: &str, job_token: &str) -> AnyhowResult<()> {
     let _record_id = self.insert(
-      event_type,
+      FirehoseEvent::W2lTemplateUploadStarted,
       Some(user_token),
       Some(job_token)
     ).await?;
+    Ok(())
+  }
 
+  pub async fn enqueue_w2l_inference(&self, maybe_user_token: Option<&str>, job_token: &str) -> AnyhowResult<()> {
+    let _record_id = self.insert(
+      FirehoseEvent::W2lInferenceStarted,
+      maybe_user_token,
+      Some(job_token)
+    ).await?;
     Ok(())
   }
 
   pub async fn publish_w2l_template_upload_finished(&self, user_token: &str, template_token: &str) -> AnyhowResult<()> {
-    let token = random_prefix_crockford_token("EV", 32)?;
-    let event_type = FirehoseEvent::W2lTemplateUploadCompleted;
-
     let _record_id = self.insert(
-      event_type,
+    FirehoseEvent::W2lTemplateUploadCompleted,
       Some(user_token),
       Some(template_token)
     ).await?;
-
     Ok(())
   }
 
+  pub async fn w2l_inference_finished(&self, maybe_user_token: Option<&str>, job_token: &str) -> AnyhowResult<()> {
+    let _record_id = self.insert(
+      FirehoseEvent::W2lInferenceCompleted,
+      maybe_user_token,
+      Some(job_token)
+    ).await?;
+    Ok(())
+  }
+
+  // =======================================================================
+
   async fn insert(
     &self,
     event_type: FirehoseEvent,

src/http_server/endpoints/events/list_events.rs

@@ -5,7 +5,9 @@ use actix_web::dev::HttpResponseBuilder;
 use actix_web::error::ResponseError;
 use actix_web::http::StatusCode;
 use actix_web::{Responder, web, HttpResponse, error, HttpRequest, HttpMessage};
+use chrono::{DateTime, Utc};
 use crate::AnyhowResult;
+use crate::common_queries::list_w2l_templates::{list_w2l_templates, W2lTemplateRecordForList};
 use crate::common_queries::sessions::create_session_for_user;
 use crate::http_server::endpoints::users::create_account::CreateAccountError::{BadInput, ServerError, UsernameTaken, EmailTaken};
 use crate::http_server::endpoints::users::login::LoginSuccessResponse;
@@ -23,8 +25,6 @@ use sqlx::error::DatabaseError;
 use sqlx::error::Error::Database;
 use sqlx::mysql::MySqlDatabaseError;
 use std::sync::Arc;
-use chrono::{DateTime, Utc};
-use crate::common_queries::list_w2l_templates::{list_w2l_templates, W2lTemplateRecordForList};
 
 #[derive(Serialize)]
 pub struct EventRecord {

src/http_server/endpoints/w2l/enqueue_infer_w2l_with_uploads.rs

@@ -341,6 +341,14 @@ SET
   //  .map_err(|err| HttpResponse::InternalServerError()
   //    .body(format!("Redis Err: {:?}", err)))?;
 
+  server_state.firehose_publisher.enqueue_w2l_inference(maybe_user_token.as_deref(), &job_token)
+    .await
+    .map_err(|e| {
+      warn!("error publishing event: {:?}", e);
+      InferW2lWithUploadError::ServerError
+    })?;
+
+
   let response = InferW2lWithUploadSuccessResponse {
     success: true,
     job_token: job_token.to_string(),

src/http_server/endpoints/w2l/enqueue_upload_w2l_template.rs

@@ -200,14 +200,13 @@ SET
 
   info!("new w2l template upload job id: {}", record_id);
 
-  server_state.firehose_publisher.publish_w2l_template_upload_enqueue(&user_session.user_token, &job_token)
+  server_state.firehose_publisher.enqueue_w2l_template_upload(&user_session.user_token, &job_token)
     .await
     .map_err(|e| {
       warn!("error publishing event: {:?}", e);
       UploadW2lTemplateError::ServerError
     })?;
 
-
   let response = UploadW2lTemplateSuccessResponse {
     success: true,
     job_token: job_token.to_string(),

src/job_queries/w2l_inference_job_queries.rs

@@ -163,7 +163,7 @@ pub async fn insert_w2l_result<P: AsRef<Path>>(
   frame_width: u32,
   frame_height: u32,
   duration_millis: u64
-) -> AnyhowResult<u64>
+) -> AnyhowResult<(u64, String)>
 {
   let inference_result_token = random_prefix_crockford_token("W2L_RES:", 32)?;
 
@@ -216,7 +216,7 @@ SET
     }
   };
 
-  Ok(record_id)
+  Ok((record_id, inference_result_token.clone()))
 }
 
 pub struct W2lTemplateRecord2 {

src/w2l_inference_job.rs

@@ -8,6 +8,7 @@
 #[macro_use] extern crate serde_derive;
 
 pub mod buckets;
+pub mod common_queries;
 pub mod job_queries;
 pub mod script_execution;
 pub mod util;
@@ -18,6 +19,7 @@ use crate::buckets::bucket_client::BucketClient;
 use crate::buckets::bucket_path_unifier::BucketPathUnifier;
 use crate::buckets::bucket_paths::hash_to_bucket_path;
 use crate::buckets::file_hashing::get_file_hash;
+use crate::common_queries::firehose_publisher::FirehosePublisher;
 use crate::job_queries::w2l_inference_job_queries::W2lInferenceJobRecord;
 use crate::job_queries::w2l_inference_job_queries::get_w2l_template_by_token;
 use crate::job_queries::w2l_inference_job_queries::insert_w2l_result;
@@ -43,9 +45,9 @@ use std::fs::{File, metadata};
 use std::io::{BufReader, Read};
 use std::path::{PathBuf, Path};
 use std::process::Command;
+use std::thread;
 use std::time::Duration;
 use tempdir::TempDir;
-use std::thread;
 
 // Buckets (shared config)
 const ENV_ACCESS_KEY : &'static str = "ACCESS_KEY";
@@ -77,6 +79,8 @@ struct Inferencer {
   pub private_bucket_client: BucketClient,
   pub public_bucket_client: BucketClient,
 
+  pub firehose_publisher: FirehosePublisher,
+
   pub bucket_path_unifier: BucketPathUnifier,
   pub semi_persistent_cache: SemiPersistentCacheDir,
 
@@ -190,6 +194,10 @@ async fn main() -> AnyhowResult<()> {
   let w2l_end_bump_filename = easyenv::get_env_string_or_default(
     "W2L_END_BUMP_FILENAME", "vocodes-short-end-bump.mp4");
 
+  let firehose_publisher = FirehosePublisher {
+    mysql_pool: mysql_pool.clone(), // NB: MySqlPool is clone/send/sync safe
+  };
+
   let inferencer = Inferencer {
     download_temp_directory: temp_directory,
     mysql_pool,
@@ -203,6 +211,7 @@ async fn main() -> AnyhowResult<()> {
     w2l_inference: w2l_inference_command,
     bucket_path_unifier: BucketPathUnifier::default_paths(),
     semi_persistent_cache,
+    firehose_publisher,
     w2l_model_filename,
     w2l_end_bump_filename,
   };
@@ -510,7 +519,7 @@ async fn process_job(inferencer: &Inferencer, job: &W2lInferenceJobRecord) -> An
   // ==================== SAVE RECORDS ==================== //
 
   info!("Saving w2l inference record...");
-  let id = insert_w2l_result(
+  let (id, inference_result_token) = insert_w2l_result(
     &inferencer.mysql_pool,
     job,
     &result_object_path,
@@ -524,6 +533,15 @@ async fn process_job(inferencer: &Inferencer, job: &W2lInferenceJobRecord) -> An
   info!("Marking job complete...");
   mark_w2l_inference_job_done(&inferencer.mysql_pool, job, true).await?;
 
+  inferencer.firehose_publisher.w2l_inference_finished(
+    job.maybe_creator_user_token.as_deref(),
+    &inference_result_token)
+    .await
+    .map_err(|e| {
+      warn!("error publishing event: {:?}", e);
+      anyhow!("error publishing event")
+    })?;
+
   info!("Job {} complete success! Downloaded, ran inference, and uploaded. Saved model record: {}",
         job.id, id);