Merge pull request #144 from storytold/brandon-email-job

email-job

Cargo.lock

@@ -2338,6 +2338,56 @@ dependencies = [
  "memchr",
 ]
 
+[[package]]
+name = "email-sender-job"
+version = "0.0.1"
+dependencies = [
+ "actix-web 4.0.1",
+ "actix_helpers",
+ "anyhow",
+ "async-trait",
+ "bootstrap",
+ "buckets",
+ "chrono",
+ "clap 2.34.0",
+ "cloud_storage",
+ "concurrency",
+ "config",
+ "container_common",
+ "dotenv",
+ "easyenv",
+ "email_sender",
+ "enums",
+ "errors",
+ "filesys",
+ "hashing",
+ "hostname",
+ "hyper",
+ "images",
+ "jobs_common",
+ "lfu",
+ "log",
+ "media",
+ "memory_caching",
+ "mimetypes",
+ "mysql_queries",
+ "once_cell",
+ "r2d2_redis",
+ "redis_common",
+ "serde",
+ "serde_derive",
+ "serde_json",
+ "server_environment",
+ "sqlx",
+ "subprocess",
+ "subprocess_common",
+ "tempdir",
+ "testing",
+ "tokens",
+ "tokio 1.34.0",
+ "tts_common",
+]
+
 [[package]]
 name = "email_address"
 version = "0.2.4"
@@ -2381,9 +2431,11 @@ dependencies = [
  "my-workspace-hack",
  "serde",
  "serde_derive",
+ "serde_json",
  "sqlx",
  "sqlx-core",
  "strum",
+ "toml",
 ]
 
 [[package]]

Cargo.toml

@@ -68,6 +68,7 @@ members = [
   # Services: API servers, jobs, workers, etc.
   "crates/service/job/analytics_job",
   "crates/service/job/download_job",
+  "crates/service/job/email_sender_job",
   "crates/service/job/inference_job",
   "crates/service/job/tts_download_job",
   "crates/service/job/tts_inference_job",

_sql/migrations/2023-11-15-120921_email_sender_jobs/down.sql

@@ -0,0 +1,5 @@
+-- noinspection SqlDialectInspectionForFile
+-- noinspection SqlNoDataSourceInspectionForFile
+-- noinspection SqlResolveForFile
+
+DROP TABLE email_sender_jobs;

_sql/migrations/2023-11-15-120921_email_sender_jobs/up.sql

@@ -0,0 +1,160 @@
+-- noinspection SqlDialectInspectionForFile
+-- noinspection SqlNoDataSourceInspectionForFile
+-- noinspection SqlResolveForFile
+
+CREATE TABLE email_sender_jobs (
+
+  -- Not used for anything except replication.
+  id BIGINT(20) NOT NULL AUTO_INCREMENT,
+
+  -- Effective "primary key" (PUBLIC)
+  -- This is so the in-progress results can be looked up by the UI.
+  token VARCHAR(32) NOT NULL,
+
+  -- Idempotency token from client
+  -- This is so we don't submit duplicate jobs.
+  uuid_idempotency_token VARCHAR(36) NOT NULL,
+
+  -- ========== EMAIL TYPE AND ARGS ==========
+
+  -- Type of email (and the template) to use.
+  -- This is the most important part of the email job.
+  -- Possible values:
+  --  * password_reset
+  --  * welcome
+  email_category VARCHAR(32) NOT NULL,
+
+  -- Polymorphic arguments payload that depends on the type of email job.
+  -- TEXT = 65,535 bytes (64 KiB), ~= 4 bytes per UTF-8 character, ~ 16383 characters.
+  maybe_email_args TEXT DEFAULT NULL,
+
+  -- ========== EMAIL DESTINATION ==========
+
+  -- Email we're sending to (always authoritative over the user token
+  -- record, even if user token is set)
+  destination_email_address VARCHAR(255) DEFAULT NULL,
+
+  -- Maybe the user record this is being sent to.
+  maybe_destination_user_token  VARCHAR(32) DEFAULT NULL,
+
+  -- ========== EMAIL LANGUAGE ==========
+
+  -- Email the language should be sent in.
+  -- The full IETF BCP47 language tag (eg. en, en-US, es-419, ja-JP, pt, etc.)
+  -- The wide varchar space is because these language tags can become cumbersome, eg "hy-Latn-IT-arevela"
+  ietf_language_tag VARCHAR(64) NOT NULL DEFAULT 'en',
+
+  -- Email the language should be sent in.
+  -- The IETF BCP47 language tag's primary language subtag (eg. "es-419" becomes "es" and "en-US" becomes "en")
+  ietf_primary_language_subtag VARCHAR(12) NOT NULL DEFAULT 'en',
+
+  -- ========== CREATOR DETAILS ==========
+
+  -- Foreign key to user that triggered the job (eg. for user DMs)
+  maybe_creator_user_token VARCHAR(32) DEFAULT NULL,
+
+  -- AVT of the user that triggered the job (eg. for user DMs)
+  -- Based on a cookie sent by the frontend.
+  -- We'll save this even if the user is logged in.
+  maybe_creator_anonymous_visitor_token VARCHAR(32) DEFAULT NULL,
+
+  -- For abuse tracking.
+  -- Wide enough for IPv4/6
+  maybe_creator_ip_address VARCHAR(40) DEFAULT NULL,
+
+  -- ========== WORKLOAD SOURCE AND PRIORITY ==========
+
+  -- Priority *increases*, so a level of 2 will be higher than 1.
+  -- These are the level semantics currently:
+  --   - (N/A; all emails should be roughly the same importance for now)
+  priority_level SMALLINT UNSIGNED NOT NULL DEFAULT 0,
+
+  -- ========== DEVELOPMENT AND DEBUGGING METADATA ==========
+
+  -- If true, the request gets a "debug" flag, which may do different
+  -- things depending on the type of work. This doesn't really have any
+  -- specific meaning -- it can be used to flag for anything -- but should
+  -- typically be used for one-off debugging and not a part of the normal
+  -- product surface area. Contrast this with `maybe_routing_tag`, which
+  -- forces the job onto a specific physical server: this might be a flag
+  -- deployed to handle debugging across the entire fleet.
+  is_debug_request BOOLEAN NOT NULL DEFAULT FALSE,
+
+  -- If set, the request gets processed by a special "tagged" worker
+  -- that matches this tag. The ordinary workers will ignore this work.
+  -- The tag is typically added via a special HTTP header when the work
+  -- is enqueued. In practice, this means we can canary deploy or cordon
+  -- off special workers to handle certain jobs.
+  maybe_routing_tag VARCHAR(32) DEFAULT NULL,
+
+  -- ========== JOB SYSTEM DETAILS ==========
+
+  -- Jobs begin as "pending", then transition to other states.
+  --
+  --  * Pending = job is ready to go
+  --  * Started = job is running
+  --  * Complete_Success = job is done (success)
+  --  * Complete_Failure = job is done (permanent failure, eg. a Face Animator job without a face)
+  --  * Attempt_Failed = job failed but may retry.
+  --  * Dead = job failed permanently (ie. exhausted retries)
+  --  * Cancelled_By_User = the user canceled their own job
+  --  * Cancelled_By_System = the system terminated the job (ie. load shedding, user account deletion, etc.)
+  --
+  -- Pending -> Started -> Complete_Success
+  --                    |-> Complete_Failure
+  --                    \-> Attempt_Failed -> Started -> { Complete, Failed, Dead }
+  status ENUM(
+    'pending',
+    'started',
+    'complete_success',
+    'complete_failure',
+    'attempt_failed',
+    'dead',
+    'cancelled_by_user',
+    'cancelled_by_system'
+  ) NOT NULL DEFAULT 'pending',
+
+  -- We can track this against a "max_attempt_count"
+  attempt_count SMALLINT UNSIGNED NOT NULL DEFAULT 0,
+
+  -- Optional internal-only debugging information in the case of failure.
+  internal_debugging_failure_reason VARCHAR(512) DEFAULT NULL,
+
+  -- Worker hostname (linux hostname, k8s pod name)
+  -- Assigned when a worker picks up the job
+  -- Reassigned if the job fails and gets picked up again
+  assigned_worker VARCHAR(128) DEFAULT NULL,
+
+  -- Cluster name (k8s)
+  -- Assigned when a worker picks up the job
+  -- Reassigned if the job fails and gets picked up again
+  assigned_cluster VARCHAR(128) DEFAULT NULL,
+
+  created_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP,
+  updated_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
+
+  -- Failed jobs will set a next attempt time.
+  -- Subsequent tries can increase the timeout.
+  -- Failures because of permissions require human intervention => [retry_at=null].
+  -- Failures because of invalid files are dead => [status=dead].
+  retry_at TIMESTAMP NULL,
+
+  -- Set when the job first starts executing
+  first_started_at TIMESTAMP NULL,
+
+  -- Set when the job is successfully completed
+  successfully_completed_at TIMESTAMP NULL,
+
+  -- INDICES --
+  PRIMARY KEY (id),
+  UNIQUE KEY (token),
+  UNIQUE KEY (uuid_idempotency_token),
+  KEY index_email_category (email_category),
+  KEY fk_maybe_creator_user_token (maybe_creator_user_token),
+  KEY index_maybe_creator_ip_address (maybe_creator_ip_address),
+  KEY index_priority_level (priority_level),
+  KEY index_is_debug_request (is_debug_request),
+  KEY index_maybe_routing_tag (maybe_routing_tag),
+  KEY index_status (status)
+
+) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_bin;

crates/_types/enums/Cargo.toml

@@ -29,3 +29,7 @@ my-workspace-hack = { version = "0.1", path = "../../../my-workspace-hack" }
 
 # Enum utilities
 strum = { version = "0.24", features = ["derive"] }
+
+# Test serialization
+serde_json = "1.0.64"
+toml = "0.5.8"

crates/_types/enums/src/by_table/email_sender_jobs/email_category.rs

@@ -0,0 +1,115 @@
+use std::collections::BTreeSet;
+
+#[cfg(test)]
+use strum::EnumCount;
+#[cfg(test)]
+use strum::EnumIter;
+
+/// Used in the `email_sender_jobs` table in `VARCHAR(32)` field `id_category`.
+///
+/// This denotes the type of email being sent.
+///
+/// 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 EmailCategory {
+  /// User is recently registered
+  #[serde(rename = "welcome")]
+  Welcome,
+
+  /// User is resetting their password
+  #[serde(rename = "password_reset")]
+  PasswordReset,
+}
+
+// TODO(bt, 2022-12-21): This desperately needs MySQL integration tests!
+impl_enum_display_and_debug_using_to_str!(EmailCategory);
+impl_mysql_enum_coders!(EmailCategory);
+
+/// NB: Legacy API for older code.
+impl EmailCategory {
+  pub fn to_str(&self) -> &'static str {
+    match self {
+      Self::Welcome => "welcome",
+      Self::PasswordReset => "password_reset",
+    }
+  }
+
+  pub fn from_str(value: &str) -> Result<Self, String> {
+    match value {
+      "welcome" => Ok(Self::Welcome),
+      "password_reset" => Ok(Self::PasswordReset),
+      _ => 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::Welcome,
+      Self::PasswordReset,
+    ])
+  }
+}
+
+#[cfg(test)]
+mod tests {
+  use crate::by_table::email_sender_jobs::email_category::EmailCategory;
+  use crate::test_helpers::assert_serialization;
+
+  mod explicit_checks {
+    use super::*;
+
+    #[test]
+    fn test_serialization() {
+      assert_serialization(EmailCategory::Welcome, "welcome");
+      assert_serialization(EmailCategory::PasswordReset, "password_reset");
+    }
+
+    #[test]
+    fn to_str() {
+      assert_eq!(EmailCategory::Welcome.to_str(), "welcome");
+      assert_eq!(EmailCategory::PasswordReset.to_str(), "password_reset");
+    }
+
+    #[test]
+    fn from_str() {
+      assert_eq!(EmailCategory::from_str("welcome").unwrap(), EmailCategory::Welcome);
+      assert_eq!(EmailCategory::from_str("password_reset").unwrap(), EmailCategory::PasswordReset);
+    }
+
+    #[test]
+    fn all_variants() {
+      // Static check
+      let mut variants = EmailCategory::all_variants();
+      assert_eq!(variants.len(), 2);
+      assert_eq!(variants.pop_first(), Some(EmailCategory::Welcome));
+      assert_eq!(variants.pop_first(), Some(EmailCategory::PasswordReset));
+      assert_eq!(variants.pop_first(), None);
+
+      // Generated check
+      use strum::IntoEnumIterator;
+      assert_eq!(EmailCategory::all_variants().len(), EmailCategory::iter().len());
+    }
+  }
+
+  mod mechanical_checks {
+    use super::*;
+
+    #[test]
+    fn variant_length() {
+      use strum::IntoEnumIterator;
+      assert_eq!(EmailCategory::all_variants().len(), EmailCategory::iter().len());
+    }
+
+    #[test]
+    fn round_trip() {
+      for variant in EmailCategory::all_variants() {
+        assert_eq!(variant, EmailCategory::from_str(variant.to_str()).unwrap());
+        assert_eq!(variant, EmailCategory::from_str(&format!("{}", variant)).unwrap());
+        assert_eq!(variant, EmailCategory::from_str(&format!("{:?}", variant)).unwrap());
+      }
+    }
+  }
+}

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

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

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

@@ -1,6 +1,7 @@
 // ===== MySql =====
 pub mod audit_logs;
 pub mod comments;
+pub mod email_sender_jobs;
 pub mod generic_download_jobs;
 pub mod generic_inference_jobs;
 pub mod generic_synthetic_ids;

crates/_types/enums/src/common/job_status_plus.rs

@@ -15,9 +15,8 @@ use strum::EnumIter;
 /// DO NOT CHANGE VALUES WITHOUT A MIGRATION STRATEGY.
 
 #[cfg_attr(test, derive(EnumIter, EnumCount))]
-#[derive(Clone, Copy, PartialEq, Eq, Hash, Ord, PartialOrd, Serialize, Deserialize, sqlx::Type)]
+#[derive(Clone, Copy, PartialEq, Eq, Hash, Ord, PartialOrd, Serialize, Deserialize)]
 #[serde(rename_all = "snake_case")]
-#[sqlx(rename_all = "snake_case")]
 pub enum JobStatusPlus {
   Pending,
   Started,
@@ -31,6 +30,7 @@ pub enum JobStatusPlus {
 
 // TODO(bt, 2022-12-21): This desperately needs MySQL integration tests!
 impl_enum_display_and_debug_using_to_str!(JobStatusPlus);
+impl_mysql_enum_coders!(JobStatusPlus);
 
 // NB: We can derive `sqlx::Type` instead of using `impl_mysql_enum_coders`
 

crates/_types/tokens/src/prefixes.rs

@@ -26,6 +26,7 @@ pub(crate) enum TokenPrefix {
   AuditLog,
   Comment,
   DownloadJob,
+  EmailSenderJob,
   InferenceJob,
   MediaFile,
   MediaUpload,
@@ -90,6 +91,7 @@ impl PrefixGenerator for TokenPrefix {
       Self::AuditLog => "audit_",
       Self::Comment => "comment_",
       Self::DownloadJob => "jdown_", // NB: Previously "JGUP:"
+      Self::EmailSenderJob => "email_job_",
       Self::InferenceJob => "jinf_",
       Self::MediaFile => "m_",
       Self::MediaUpload => "mu_",
@@ -289,10 +291,11 @@ mod tests {
           .all(|prefix| prefix.ends_with(":") || prefix.ends_with("_")));
     }
 
+    // TODO: Just kill this test
     #[test]
     fn test_all_prefixes_end_with_separator_length_one() {
       for prefix in get_all_prefixes().iter().map(|s| *s) {
-        if prefix == "news_story_" || prefix == "tts_task_" || prefix == "INT_API:" || prefix == "pw_reset_" {
+        if prefix == "news_story_" || prefix == "tts_task_" || prefix == "INT_API:" || prefix == "pw_reset_" || prefix == "email_job_" {
           // TODO/FIXME: I'm too tired at 5AM to replacen from the left. Make this test valid.
           //  These tokens are from the AIChatBot sidecar, so asserting their validity is less important.
           continue;

crates/_types/tokens/src/tokens/email_sender_job_token.rs

@@ -0,0 +1,14 @@
+use std::fmt::Debug;
+
+use serde::Deserialize;
+use serde::Serialize;
+
+use crate::prefixes::TokenPrefix;
+
+/// The primary key for email_sender_jobs
+#[derive(Clone, PartialEq, Eq, PartialOrd, Ord, Hash, sqlx::Type, Debug, Serialize, Deserialize)]
+#[sqlx(transparent)]
+pub struct EmailSenderJobToken(pub String);
+
+impl_string_token!(EmailSenderJobToken);
+impl_crockford_generator!(EmailSenderJobToken, 32usize, TokenPrefix::EmailSenderJob, CrockfordLower);

crates/_types/tokens/src/tokens/mod.rs

@@ -3,6 +3,7 @@ pub mod api_tokens_external;
 pub mod api_tokens_internal;
 pub mod audit_logs;
 pub mod comments;
+pub mod email_sender_job_token;
 pub mod firehose_entries;
 pub mod generic_download_jobs;
 pub mod generic_inference_jobs;

crates/lib/mysql_queries/src/payloads/email_sender_jobs/email_sender_job_args.rs

@@ -0,0 +1,39 @@
+
+// TODO: Switch to protobuf or nom to better support forward-compatible changes
+
+use errors::AnyhowResult;
+
+use crate::payloads::email_sender_jobs::subtypes::email_job_password_reset_args::EmailJobPasswordResetArgs;
+use crate::payloads::email_sender_jobs::subtypes::email_job_welcome_args::EmailJobWelcomeArgs;
+
+/// Used to encode extra state for the `email_sender_jobs` table.
+/// This should act somewhat like a serialized protobuf stored inside a record.
+#[derive(Clone, Debug, Serialize, Deserialize)]
+pub struct EmailSenderJobArgs {
+  /// REQUIRED.
+  /// Actual type-specific arguments.
+  #[serde(rename = "a")] // NB: DO NOT CHANGE. It could break live jobs. Renamed to consume fewer bytes
+  #[serde(skip_serializing_if = "Option::is_none")]
+  pub args: Option<PolymorphicEmailSenderJobArgs>,
+}
+
+// NB: Keep the enum variant names short as these get serialized to json in the db
+#[derive(Clone, Debug, Serialize, Deserialize)]
+pub enum PolymorphicEmailSenderJobArgs {
+  /// Welcome email
+  W(EmailJobWelcomeArgs),
+
+  /// Forgot password / password reset args
+  Pr(EmailJobPasswordResetArgs),
+}
+
+impl EmailSenderJobArgs {
+
+  pub fn from_json(json: &str) -> AnyhowResult<Self> {
+    Ok(serde_json::from_str(json)?)
+  }
+
+  pub fn to_json(&self) -> AnyhowResult<String> {
+    Ok(serde_json::to_string(self)?)
+  }
+}

crates/lib/mysql_queries/src/payloads/email_sender_jobs/mod.rs

@@ -0,0 +1,2 @@
+pub mod email_sender_job_args;
+pub mod subtypes;

crates/lib/mysql_queries/src/payloads/email_sender_jobs/subtypes/email_job_password_reset_args.rs

@@ -0,0 +1,29 @@
+
+#[derive(Clone, Debug, Serialize, Deserialize, PartialEq, Eq)]
+pub struct EmailJobPasswordResetArgs {
+  /// REQUIRED.
+  /// The password reset code
+  #[serde(rename = "k")] // NB: DO NOT CHANGE. It could break live jobs. Renamed to consume fewer bytes
+  #[serde(skip_serializing_if = "Option::is_none")]
+  pub password_reset_secret_key: Option<String>,
+}
+
+#[cfg(test)]
+mod tests {
+  use crate::payloads::email_sender_jobs::subtypes::email_job_password_reset_args::EmailJobPasswordResetArgs;
+
+  fn assert_json_deserializes_to_match(json: &str, original: &EmailJobPasswordResetArgs) {
+    let duplicate: EmailJobPasswordResetArgs = serde_json::de::from_str(json).unwrap();
+    assert_eq!(&duplicate, original);
+  }
+
+  #[test]
+  fn test_secret_key() {
+    let args = EmailJobPasswordResetArgs {
+      password_reset_secret_key: Some("code".to_string()),
+    };
+    let json = serde_json::ser::to_string(&args).unwrap();
+    assert_eq!(json, r#"{"k":"code"}"#.to_string());
+    assert_json_deserializes_to_match(&json, &args);
+  }
+}

crates/lib/mysql_queries/src/payloads/email_sender_jobs/subtypes/email_job_welcome_args.rs

@@ -0,0 +1,5 @@
+
+#[derive(Clone, Debug, Serialize, Deserialize)]
+pub struct EmailJobWelcomeArgs {
+
+}

crates/lib/mysql_queries/src/payloads/email_sender_jobs/subtypes/mod.rs

@@ -0,0 +1,2 @@
+pub mod email_job_password_reset_args;
+pub mod email_job_welcome_args;

crates/lib/mysql_queries/src/payloads/mod.rs

@@ -1,4 +1,5 @@
 //! This module is used to serialize values to database 'TEXT' fields.
 
+pub mod email_sender_jobs;
 pub mod generic_inference_args;
 pub mod media_upload_modification_details;

crates/lib/mysql_queries/src/queries/email_sender_jobs/_keys.rs

@@ -0,0 +1,6 @@
+//! Primary and foreign keys
+
+/// Since jobs are internal, we'll use IDs as primary keys.
+#[derive(Clone, Copy, PartialOrd, Ord, PartialEq, Eq, Debug, Serialize, Deserialize, sqlx::Type)]
+#[sqlx(transparent)]
+pub struct EmailSenderJobId(pub i64);

crates/lib/mysql_queries/src/queries/email_sender_jobs/insert_email_sender_job.rs

@@ -0,0 +1,112 @@
+use anyhow::anyhow;
+use sqlx::MySqlPool;
+
+use enums::by_table::email_sender_jobs::email_category::EmailCategory;
+use errors::AnyhowResult;
+use tokens::tokens::anonymous_visitor_tracking::AnonymousVisitorTrackingToken;
+use tokens::tokens::email_sender_job_token::EmailSenderJobToken;
+use tokens::tokens::users::UserToken;
+
+use crate::payloads::email_sender_jobs::email_sender_job_args::EmailSenderJobArgs;
+
+pub struct InsertEmailSenderJobArgs<'a> {
+  pub uuid_idempotency_token: &'a str,
+
+  pub destination_email_address: &'a str,
+  pub maybe_destination_user_token: Option<&'a UserToken>,
+
+  pub email_category: EmailCategory,
+  pub maybe_email_args: Option<EmailSenderJobArgs>,
+
+  pub ietf_language_tag: &'a str,
+  pub ietf_primary_language_subtag: &'a str,
+
+  pub maybe_creator_user_token: Option<&'a UserToken>,
+  pub maybe_avt_token: Option<&'a AnonymousVisitorTrackingToken>,
+  pub creator_ip_address: &'a str,
+
+  pub priority_level: u8,
+
+  pub is_debug_request: bool,
+  pub maybe_routing_tag: Option<&'a str>,
+
+  pub mysql_pool: &'a MySqlPool,
+}
+
+pub async fn insert_email_sender_job(args: InsertEmailSenderJobArgs<'_>) -> AnyhowResult<(EmailSenderJobToken, u64)> {
+  let job_token = EmailSenderJobToken::generate();
+
+  let serialized_args_payload = serde_json::ser::to_string(&args.maybe_email_args)
+      .map_err(|_e| anyhow!("could not encode inference args"))?;
+
+  // The routing tag column is VARCHAR(32), so we should truncate.
+  let maybe_routing_tag = args.maybe_routing_tag
+      .map(|routing_tag| {
+        let mut routing_tag = routing_tag.trim().to_string();
+        routing_tag.truncate(32);
+        routing_tag
+      });
+
+  let query = sqlx::query!(
+        r#"
+INSERT INTO email_sender_jobs
+SET
+  token = ?,
+  uuid_idempotency_token = ?,
+
+  destination_email_address = ?,
+  maybe_destination_user_token = ?,
+
+  email_category = ?,
+  maybe_email_args = ?,
+
+  ietf_language_tag = ?,
+  ietf_primary_language_subtag = ?,
+
+  maybe_creator_user_token = ?,
+  maybe_creator_anonymous_visitor_token = ?,
+  maybe_creator_ip_address = ?,
+
+  priority_level = ?,
+
+  is_debug_request = ?,
+  maybe_routing_tag = ?,
+
+  status = "pending"
+        "#,
+        job_token.as_str(),
+        args.uuid_idempotency_token,
+
+        args.destination_email_address,
+        args.maybe_destination_user_token,
+
+        args.email_category,
+        serialized_args_payload,
+
+        args.ietf_language_tag,
+        args.ietf_primary_language_subtag,
+
+        args.maybe_creator_user_token.map(|t| t.to_string()),
+        args.maybe_avt_token.map(|t| t.to_string()),
+        args.creator_ip_address,
+
+        args.priority_level,
+
+        args.is_debug_request,
+        maybe_routing_tag,
+    );
+
+  let query_result = query.execute(args.mysql_pool)
+      .await;
+
+  let record_id = match query_result {
+    Ok(res) => {
+      res.last_insert_id()
+    },
+    Err(err) => {
+      return Err(anyhow!("error inserting new generic inference job: {:?}", err));
+    }
+  };
+
+  Ok((job_token, record_id))
+}

crates/lib/mysql_queries/src/queries/email_sender_jobs/list_available_email_sender_jobs.rs

@@ -0,0 +1,277 @@
+use anyhow::anyhow;
+use chrono::Utc;
+use sqlx::MySqlPool;
+
+use enums::by_table::email_sender_jobs::email_category::EmailCategory;
+use enums::common::job_status_plus::JobStatusPlus;
+use errors::AnyhowResult;
+use tokens::tokens::anonymous_visitor_tracking::AnonymousVisitorTrackingToken;
+use tokens::tokens::email_sender_job_token::EmailSenderJobToken;
+use tokens::tokens::users::UserToken;
+
+use crate::helpers::boolean_converters::i8_to_bool;
+use crate::payloads::email_sender_jobs::email_sender_job_args::EmailSenderJobArgs;
+use crate::queries::email_sender_jobs::_keys::EmailSenderJobId;
+
+/// table: email_sender_jobs
+#[derive(Debug)]
+pub struct AvailableEmailSenderJob {
+  pub id: EmailSenderJobId,
+  pub token: EmailSenderJobToken,
+
+  // Job type
+  pub email_category: EmailCategory,
+  pub maybe_email_args: Option<EmailSenderJobArgs>,
+
+  // Email details
+  pub destination_email_address: String,
+  pub maybe_destination_user_token: Option<UserToken>,
+
+  // Language details
+  pub ietf_language_tag: Option<String>,
+  pub ietf_primary_language_subtag: Option<String>,
+
+  // User information to propagate downstream
+  pub maybe_creator_user_token: Option<UserToken>,
+  pub maybe_creator_anonymous_visitor_token: Option<AnonymousVisitorTrackingToken>,
+  pub maybe_creator_ip_address: String,
+
+  // Job information
+  pub status: JobStatusPlus,
+  pub attempt_count: u16,
+
+  pub priority_level: u16,
+
+  // Development / debug info
+  pub is_debug_request: bool,
+  pub maybe_routing_tag: Option<String>,
+
+  pub created_at: chrono::DateTime<Utc>,
+  pub updated_at: chrono::DateTime<Utc>,
+  pub retry_at: Option<chrono::DateTime<Utc>>,
+}
+
+pub struct ListAvailableEmailSenderJobArgs<'a> {
+  pub num_records: u32,
+  pub is_debug_worker: bool,
+  pub sort_by_priority: bool,
+  pub mysql_pool: &'a MySqlPool,
+}
+
+pub async fn list_available_email_sender_jobs(
+  args: ListAvailableEmailSenderJobArgs<'_>,
+)
+  -> AnyhowResult<Vec<AvailableEmailSenderJob>>
+{
+  let query = if args.sort_by_priority {
+    list_sorted_by_priority(args).await
+  } else {
+    list_sorted_by_id(args).await
+  };
+
+  let job_records = query?;
+
+  let job_records : Vec<AvailableEmailSenderJob> = job_records.into_iter()
+      .map(|record : AvailableEmailSenderJobRawInternal| {
+        let record = AvailableEmailSenderJob {
+          id: EmailSenderJobId(record.id),
+          token: EmailSenderJobToken::new_from_str(&record.token),
+          email_category: EmailCategory::from_str(&record.email_category)
+              .map_err(|err| anyhow!("incorrect category: {:?}", err))?,
+          maybe_email_args: record.maybe_email_args
+              .as_deref()
+              .map(|args| EmailSenderJobArgs::from_json(args))
+              .transpose()?,
+          destination_email_address: record.destination_email_address,
+          maybe_destination_user_token: record.maybe_destination_user_token
+              .map(|token| UserToken::new_from_str(&token)),
+          ietf_language_tag: record.ietf_language_tag,
+          ietf_primary_language_subtag: record.ietf_primary_language_subtag,
+          maybe_creator_ip_address: record.maybe_creator_ip_address,
+          maybe_creator_user_token: record.maybe_creator_user_token
+              .map(|token| UserToken::new_from_str(&token)),
+          maybe_creator_anonymous_visitor_token: record.maybe_creator_anonymous_visitor_token
+              .map(|token| AnonymousVisitorTrackingToken::new_from_str(&token)),
+          status: JobStatusPlus::from_str(&record.status)
+              .map_err(|err| anyhow!("incorrect category: {:?}", err))?,
+          attempt_count: record.attempt_count,
+          priority_level: record.priority_level,
+          is_debug_request: i8_to_bool(record.is_debug_request),
+          maybe_routing_tag: record.maybe_routing_tag,
+          created_at: record.created_at,
+          updated_at: record.updated_at,
+          retry_at: record.retry_at,
+        };
+        Ok(record)
+      })
+      // NB: Magic Vec<Result> -> Result<Vec<>>
+      // https://stackoverflow.com/a/63798748
+      .into_iter()
+      .collect::<Result<Vec<AvailableEmailSenderJob>, anyhow::Error>>()?;
+
+  Ok(job_records)
+}
+
+async fn list_sorted_by_id(args: ListAvailableEmailSenderJobArgs<'_>) -> Result<Vec<AvailableEmailSenderJobRawInternal>, sqlx::Error> {
+  // NB: Can't be type checked because of WHERE IN clause with dynamic contents
+  // TODO(bt,2023-11-15): Include the types. This should be a static query, not a query builder.
+
+  let mut query = r#"
+SELECT
+  id,
+  token,
+
+  email_category,
+  maybe_email_args,
+
+  destination_email_address,
+  maybe_destination_user_token,
+
+  ietf_language_tag,
+  ietf_primary_language_subtag,
+
+  maybe_creator_user_token,
+  maybe_creator_anonymous_visitor_token,
+  maybe_creator_ip_address,
+
+  status,
+  attempt_count,
+
+  priority_level,
+
+  is_debug_request,
+  maybe_routing_tag,
+
+  created_at,
+  updated_at,
+  retry_at
+
+FROM email_sender_jobs"#.to_string();
+
+  query.push_str(r#"
+  WHERE
+  (
+    status IN ("pending", "attempt_failed")
+  )
+  AND
+  (
+    retry_at IS NULL
+    OR
+    retry_at < CURRENT_TIMESTAMP
+  )
+  AND
+  (
+    is_debug_request = ?
+  )
+  ORDER BY id ASC
+  LIMIT ?
+        "#);
+
+  let mut query = sqlx::query_as::<_, AvailableEmailSenderJobRawInternal>(&query)
+      .bind(args.is_debug_worker)
+      .bind(args.num_records);
+
+  query.fetch_all(args.mysql_pool)
+      .await
+}
+
+async fn list_sorted_by_priority(args: ListAvailableEmailSenderJobArgs<'_>) -> Result<Vec<AvailableEmailSenderJobRawInternal>, sqlx::Error> {
+  // NB: Can't be type checked because of WHERE IN clause with dynamic contents
+  // TODO(bt,2023-11-15): Include the types. This should be a static query, not a query builder.
+
+  let mut query = r#"
+SELECT
+  id,
+  token,
+
+  email_category,
+  maybe_email_args,
+
+  destination_email_address,
+  maybe_destination_user_token,
+
+  ietf_language_tag,
+  ietf_primary_language_subtag,
+
+  maybe_creator_user_token,
+  maybe_creator_anonymous_visitor_token,
+  maybe_creator_ip_address,
+
+  status,
+  attempt_count,
+
+  priority_level,
+
+  is_debug_request,
+  maybe_routing_tag,
+
+  created_at,
+  updated_at,
+  retry_at
+
+FROM email_sender_jobs"#.to_string();
+
+  query.push_str(r#"
+  WHERE
+  (
+    status IN ("pending", "attempt_failed")
+  )
+  AND
+  (
+    retry_at IS NULL
+    OR
+    retry_at < CURRENT_TIMESTAMP
+  )
+  AND
+  (
+    is_debug_request = ?
+  )
+  ORDER BY priority_level DESC, id ASC
+  LIMIT ?
+        "#);
+
+  let mut query = sqlx::query_as::<_, AvailableEmailSenderJobRawInternal>(&query)
+      .bind(args.is_debug_worker)
+      .bind(args.num_records);
+
+  query.fetch_all(args.mysql_pool)
+      .await
+}
+
+#[derive(Debug)]
+#[derive(sqlx::FromRow)]
+struct AvailableEmailSenderJobRawInternal {
+  pub id: i64,
+  pub token: String,
+
+  // Job type
+  pub email_category: String,
+  pub maybe_email_args: Option<String>,
+
+  // Email details
+  pub destination_email_address: String,
+  pub maybe_destination_user_token: Option<String>,
+
+  // Language details
+  pub ietf_language_tag: Option<String>,
+  pub ietf_primary_language_subtag: Option<String>,
+
+  // User information to propagate downstream
+  pub maybe_creator_user_token: Option<String>,
+  pub maybe_creator_anonymous_visitor_token: Option<String>,
+  pub maybe_creator_ip_address: String,
+
+  // Job information
+  pub status: String,
+  pub attempt_count: u16,
+
+  pub priority_level: u16,
+
+  // Development / debug info
+  pub is_debug_request: i8,
+  pub maybe_routing_tag: Option<String>,
+
+  pub created_at: chrono::DateTime<Utc>,
+  pub updated_at: chrono::DateTime<Utc>,
+  pub retry_at: Option<chrono::DateTime<Utc>>,
+}

crates/lib/mysql_queries/src/queries/email_sender_jobs/mark_email_sender_job_completely_failed.rs

@@ -0,0 +1,41 @@
+use anyhow::anyhow;
+use sqlx::MySqlPool;
+
+use errors::AnyhowResult;
+
+use crate::queries::email_sender_jobs::list_available_email_sender_jobs::AvailableEmailSenderJob;
+
+pub async fn mark_email_sender_job_completely_failed(
+  pool: &MySqlPool,
+  job: &AvailableEmailSenderJob,
+  maybe_internal_debugging_failure_reason: Option<&str>,
+) -> AnyhowResult<()>
+{
+  // Max length of column is 512
+  let maybe_internal_debugging_failure_reason
+      = maybe_internal_debugging_failure_reason.map(|reason| {
+    let mut reason = reason.trim().to_string();
+    reason.truncate(512); // Max length of column is 512
+    reason
+  });
+
+  let query_result = sqlx::query!(
+        r#"
+UPDATE email_sender_jobs
+SET
+  status = "complete_failure",
+  internal_debugging_failure_reason = ?,
+  retry_at = NULL
+WHERE id = ?
+        "#,
+        maybe_internal_debugging_failure_reason.as_deref(),
+        job.id.0,
+    )
+      .execute(pool)
+      .await;
+
+  match query_result {
+    Err(err) => Err(anyhow!("error with query: {:?}", err)),
+    Ok(_r) => Ok(()),
+  }
+}

crates/lib/mysql_queries/src/queries/email_sender_jobs/mark_email_sender_job_failure.rs

@@ -0,0 +1,49 @@
+use anyhow::anyhow;
+use sqlx;
+use sqlx::MySqlPool;
+
+use errors::AnyhowResult;
+
+use crate::queries::email_sender_jobs::list_available_email_sender_jobs::AvailableEmailSenderJob;
+
+/// Mark a single inference job failure. The job may be re-run.
+pub async fn mark_email_sender_job_failure(
+  pool: &MySqlPool,
+  job: &AvailableEmailSenderJob,
+  internal_debugging_failure_reason: &str,
+  max_attempts: u16
+) -> AnyhowResult<()> {
+
+  // statuses: "attempt_failed", "complete_failure", "dead"
+  let mut next_status = "attempt_failed";
+
+  // Max length of column is 512
+  let mut internal_debugging_failure_reason = internal_debugging_failure_reason.trim().to_string();
+  internal_debugging_failure_reason.truncate(512);
+
+  if job.attempt_count >= max_attempts {
+    // NB: Job attempt count is incremented at start
+    next_status = "dead";
+  }
+
+  let query_result = sqlx::query!(
+        r#"
+UPDATE email_sender_jobs
+SET
+  status = ?,
+  internal_debugging_failure_reason = ?,
+  retry_at = NOW() + interval 2 minute
+WHERE id = ?
+        "#,
+        next_status,
+        &internal_debugging_failure_reason,
+        job.id.0,
+    )
+      .execute(pool)
+      .await;
+
+  match query_result {
+    Err(err) => Err(anyhow!("error with query: {:?}", err)),
+    Ok(_r) => Ok(()),
+  }
+}

crates/lib/mysql_queries/src/queries/email_sender_jobs/mark_email_sender_job_pending_and_grab_lock.rs

@@ -0,0 +1,118 @@
+use anyhow::anyhow;
+use sqlx;
+use sqlx::MySqlPool;
+
+use errors::AnyhowResult;
+
+use crate::common_inputs::container_environment_arg::ContainerEnvironmentArg;
+use crate::queries::email_sender_jobs::_keys::EmailSenderJobId;
+
+// TODO(2022-02-09): It's unclear how stuck "pending" jobs don't become perma-dead
+//  I think this logic is flawed and the job system needs to pick up stuck "pending" jobs.
+
+pub struct EmailSenderJobLockRecord {
+  id: i64,
+  status: String,
+  attempt_count: u16,
+}
+
+pub async fn mark_email_sender_job_pending_and_grab_lock(
+  pool: &MySqlPool,
+  job_id: EmailSenderJobId,
+  container_environment: &ContainerEnvironmentArg,
+) -> AnyhowResult<bool> {
+
+  // NB: We use transactions and "SELECT ... FOR UPDATE" to simulate mutexes.
+  let mut transaction = pool.begin().await?;
+
+  let maybe_record = sqlx::query_as!(
+    EmailSenderJobLockRecord,
+        r#"
+SELECT
+  id,
+  status,
+  attempt_count
+FROM email_sender_jobs
+WHERE id = ?
+FOR UPDATE
+        "#,
+        job_id.0,
+    )
+      .fetch_one(&mut transaction)
+      .await;
+
+  let record : EmailSenderJobLockRecord = match maybe_record {
+    Ok(record) => record,
+    Err(err) => {
+      match err {
+        sqlx::Error::RowNotFound => {
+          return Err(anyhow!("could not job"));
+        },
+        _ => {
+          return Err(anyhow!("query error"));
+        }
+      }
+    }
+  };
+
+  let can_transact = match record.status.as_ref() {
+    "pending" => true, // It's okay for us to take the lock.
+    "attempt_failed" => true, // We can retry.
+    "started" => false, // Job in progress (another job beat us, and we can't take the lock)
+    "complete_success" => false, // Job already complete
+    "complete_failure" => false, // Job already complete (permanently dead; no need to retry)
+    "dead" => false, // Job already complete (permanently dead; retries exhausted)
+    "cancelled_by_user" => false, // Job already complete (permanently dead; killed by user)
+    "cancelled_by_system" => false, // Job already complete (permanently dead; killed by system)
+    _ => false, // Future-proof
+  };
+
+  if !can_transact {
+    transaction.rollback().await?;
+    return Ok(false);
+  }
+
+  if record.attempt_count == 0 {
+    let _acquire_lock = sqlx::query!(
+        r#"
+UPDATE email_sender_jobs
+SET
+  status = 'started',
+  assigned_worker = ?,
+  assigned_cluster = ?,
+  attempt_count = attempt_count + 1,
+  retry_at = NOW() + interval 2 minute,
+  first_started_at = NOW()
+WHERE id = ?
+        "#,
+        &container_environment.hostname,
+        &container_environment.cluster_name,
+        job_id.0,
+    )
+        .execute(&mut transaction)
+        .await?;
+  } else {
+    let _acquire_lock = sqlx::query!(
+        r#"
+UPDATE email_sender_jobs
+SET
+  status = 'started',
+  assigned_worker = ?,
+  assigned_cluster = ?,
+  attempt_count = attempt_count + 1,
+  retry_at = NOW() + interval 2 minute
+WHERE id = ?
+        "#,
+        &container_environment.hostname,
+        &container_environment.cluster_name,
+        job_id.0,
+    )
+        .execute(&mut transaction)
+        .await?;
+  }
+
+
+  transaction.commit().await?;
+
+  Ok(true)
+}

crates/lib/mysql_queries/src/queries/email_sender_jobs/mark_email_sender_job_successfully_done.rs

@@ -0,0 +1,32 @@
+use anyhow::anyhow;
+use sqlx::MySqlPool;
+
+use errors::AnyhowResult;
+
+use crate::queries::email_sender_jobs::list_available_email_sender_jobs::AvailableEmailSenderJob;
+
+pub async fn mark_email_sender_job_successfully_done(
+  pool: &MySqlPool,
+  job: &AvailableEmailSenderJob,
+) -> AnyhowResult<()>
+{
+  let query_result = sqlx::query!(
+        r#"
+UPDATE email_sender_jobs
+SET
+  status = "complete_success",
+  internal_debugging_failure_reason = NULL,
+  retry_at = NULL,
+  successfully_completed_at = NOW()
+WHERE id = ?
+        "#,
+        job.id.0,
+    )
+      .execute(pool)
+      .await;
+
+  match query_result {
+    Err(err) => Err(anyhow!("error with query: {:?}", err)),
+    Ok(_r) => Ok(()),
+  }
+}

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

@@ -0,0 +1,7 @@
+pub mod _keys;
+pub mod insert_email_sender_job;
+pub mod list_available_email_sender_jobs;
+pub mod mark_email_sender_job_completely_failed;
+pub mod mark_email_sender_job_failure;
+pub mod mark_email_sender_job_pending_and_grab_lock;
+pub mod mark_email_sender_job_successfully_done;

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

@@ -2,6 +2,7 @@ pub mod api_tokens;
 pub mod audit_logs;
 pub mod billing;
 pub mod comments;
+pub mod email_sender_jobs;
 pub mod generic_download;
 pub mod generic_inference;
 pub mod generic_synthetic_ids;
@@ -19,5 +20,5 @@ pub mod users;
 pub mod vocoder;
 pub mod voice_clone_requests;
 pub mod voice_conversion;
-pub mod w2l;
 pub mod voice_designer;
+pub mod w2l;

crates/lib/mysql_queries/src/queries/users/user_password_resets/create_password_reset_request.rs

@@ -6,7 +6,7 @@ use tokens::tokens::password_reset::PasswordResetToken;
 
 use crate::queries::users::user::lookup_user_for_login_result::UserRecordForLogin;
 
-pub async fn create_password_reset(pool: &MySqlPool, user: &UserRecordForLogin, ip_address: String, secret_key: String) -> AnyhowResult<()> {
+pub async fn create_password_reset(pool: &MySqlPool, user: &UserRecordForLogin, ip_address: &str, secret_key: String) -> AnyhowResult<()> {
     let token = PasswordResetToken::generate();
 
     sqlx::query!(

crates/service/job/email_sender_job/Cargo.toml

@@ -0,0 +1,88 @@
+[package]
+name = "email-sender-job"
+edition = "2018"
+version = "0.0.1"
+authors = [
+    "Brandon Thomas <bt@brand.io>",
+    "Brandon Thomas <echelon@gmail.com>",
+]
+publish = false
+
+[[bin]]
+name = "email-sender-job"
+path = "src/main.rs"
+
+[dependencies]
+
+# Internal
+actix_helpers = { path = "../../../lib/actix_helpers" }
+bootstrap = { path = "../../../lib/bootstrap" }
+buckets = { path = "../../../_types/buckets" }
+cloud_storage = { path = "../../../lib/api_clients/cloud_storage" }
+concurrency = { path = "../../../lib/concurrency" }
+config = { path = "../../../lib/deprecated/config" }
+container_common = { path = "../../../lib/deprecated/container_common" }
+easyenv = { path = "../../../lib/easyenv" }
+email_sender = { path = "../../../lib/email_sender" }
+enums = { path = "../../../_types/enums" }
+errors = { path = "../../../lib/errors" }
+filesys = { path = "../../../lib/files/filesys" }
+hashing = { path = "../../../lib/files/hashing" }
+images = { path = "../../../lib/files/images" }
+jobs_common = { path = "../../../lib/jobs_common" }
+media = { path = "../../../lib/files/media" }
+memory_caching = { path = "../../../lib/caching/memory_caching" }
+mimetypes = { path = "../../../lib/files/mimetypes" }
+mysql_queries = { path = "../../../lib/mysql_queries" }
+redis_common = { path = "../../../lib/redis_common" }
+server_environment = { path = "../../../lib/server_environment" }
+subprocess_common = { path = "../../../lib/subprocess_common" }
+testing = { path = "../../../lib/testing" }
+tokens = { path = "../../../_types/tokens" }
+tts_common = { path = "../../../lib/tts_common" }
+
+# Vendored code
+# newrelic-telemetry = { path = "../../../vendor/newrelic-telemetry-sdk-rust" }
+
+# External
+anyhow = "1.0.75"
+chrono = { version = "0.4.19", features = ["serde"] }
+clap = "2.34.0" # TODO: This is really out of date, but has conflicts when updated
+dotenv = "0.15.0"
+hostname = "0.3.1"
+lfu = "0.2.5"
+log = "0.4.14"
+once_cell = "1.18.0"
+subprocess = "0.2.9"
+tempdir = "0.3.7"
+
+# MySql Database
+# "offline" supports typesafe builds without a live database
+# "rustls" doesn't appear to work with IP addresses, per:
+# https://github.com/launchbadge/sqlx/issues/1095#issuecomment-798050828
+sqlx = { version = "0.6.2", features = [ "mysql", "runtime-actix-native-tls", "offline", "chrono" ] }
+
+# Redis
+r2d2_redis = "0.14.0"
+
+# Runtime
+# NB: We switched to Tokio from `async-std` for broader ecosystem support.
+tokio = { version = "1.6.0", features = ["macros"] }
+
+# NB: Hyper is just for HTTP client use in jobs. I'd use Actix client, but I need tokio runtime
+# NB: Furthermore, the Tokio runtime is old, so I have to use an older Hyper.
+hyper = { version = "0.14.24" }
+
+# HTTP Server
+actix-web = "4.0.0-beta.10"
+
+# Serialization
+serde = "1.0.125"
+serde_derive = "1.0.125"
+serde_json = "1.0.64"
+
+# Async traits
+# NB: note: `async` trait functions are not currently supported
+# The initial round of stabilizations for the async/await language feature in Rust 1.39 did not
+# include support for async fn in traits.
+async-trait = "0.1.58"

crates/service/job/email_sender_job/config/email-sender-job.common.env

@@ -0,0 +1 @@
+# inference-job default .env configurations (these can be overridden with k8s)

crates/service/job/email_sender_job/config/email-sender-job.development.env

@@ -0,0 +1,109 @@
+# inference-job default .env configurations (these can be overridden with k8s)
+
+# NB: For my crapy internet connection
+BUCKET_TIMEOUT_SECONDS=6000
+
+# NB: To not step on storyteller-web (the main API gateway)
+HTTP_BIND_ADDRESS=0.0.0.0:11223
+
+# NB: Disable if not testing Redis
+REDIS_FOR_KEEPALIVE_URL="redis://localhost/0"
+
+# ================================= #
+#                                   #
+#            Vall-E-X               #
+#                                   #
+# ================================= #
+
+# should rename my folder 
+# abs path to the folder that contains the inference code need to get that container in the repo need basepath? relative to everyones env
+VALL_E_X_INFERENCE_ROOT_DIRECTORY="%storyteller_ml%/tts/VALL-E-X"
+#VALL_E_X_INFERENCE_ROOT_DIRECTORY = "/home/tensor/code/storyteller/storyteller-ml/tts/VALL-E-X" 
+VALL_E_X_INFERENCE_COMMAND = "python3 main.py"
+VALL_E_X_INFERENCE_MAYBE_VENV_COMMAND = "source .venv/bin/activate"
+
+# VALL_E_X_INFERENCE_MAYBE_DOCKER_IMAGE = "" # fill this out for dev?
+#VALL_E_X_INFERENCE_ROOT_DIRECTORY="%storyteller_ml%/tts/VALL-E-X"
+#VALL_E_X_INFERENCE_COMMAND="python3 main.py"
+#VALL_E_X_INFERENCE_MAYBE_VENV_COMMAND="source .env/bin/activate"
+#VALL_E_X_INFERENCE_MAYBE_DOCKER_IMAGE="" # fill this out for dev?
+
+# ================================= #
+#                                   #
+#             RVC (v2)              #
+#                                   #
+# ================================= #
+
+RVC_V2_INFERENCE_ROOT_DIRECTORY="%storyteller_ml%/voice_conversion/rvc-v2-brandon"
+RVC_V2_INFERENCE_COMMAND="python fakeyou_infer.py"
+RVC_V2_INFERENCE_MAYBE_VENV_COMMAND="source python/bin/activate"
+
+# ================================= #
+#                                   #
+#            SAD TALKER             #
+#                                   #
+# ================================= #
+
+SAD_TALKER_INFERENCE_ROOT_DIRECTORY="%storyteller_ml%/animation/SadTalker"
+SAD_TALKER_INFERENCE_COMMAND="python inference.py"
+SAD_TALKER_INFERENCE_MAYBE_VENV_COMMAND="source python/bin/activate"
+
+# ================================= #
+#                                   #
+#          FFMPEG WATERMARK         #
+#                                   #
+# ================================= #
+
+FFMPEG_LOGO_LOGO_PATH="%storyteller_rust%/includes/container_includes/image_assets/fakeyou_watermark.png"
+
+# ================================= #
+#                                   #
+#           SO-VITS-SVC             #
+#                                   #
+# ================================= #
+
+# Cache hubert in a recoverable place so do don't constantly re-download it.
+HUBERT_PATH="/tmp/hubert/checkpoint_best_legacy_500.pt"
+
+SO_VITS_SVC_INFERENCE_ROOT_DIRECTORY="/models/voice_conversion/so-vits-svc/src" #NB: so-vits-svc-paul-2
+SO_VITS_SVC_INFERENCE_MAYBE_DOCKER_IMAGE="1136fe8b50d5" # so-vits-svc-paul-2: --f0-method
+SO_VITS_SVC_INFERENCE_COMMAND="/install/python/bin/python3.10 fakeyou_infer.py" # NB: so-vits-svc-paul-2
+SO_VITS_SVC_INFERENCE_MAYBE_DEFAULT_CONFIG_PATH="/models/voice_conversion/so-vits-svc/example_config.json"
+
+# ================================= #
+#                                   #
+#         TACOTRON (LEGACY)         #
+#                                   #
+# ================================= #
+
+## Old docker image
+#TACOTRON2_VOCODES_DOCKER_IMAGE_SHA=1ba08f70f2ba
+
+TT2_LEGACY_ROOT_DIRECTORY="/models/tts"
+TT2_LEGACY_MAYBE_PYTHON_INTERPRETER="python3.6"
+TT2_LEGACY_MAYBE_DOCKER_IMAGE_SHA="9f1945388110"
+
+# ================================= #
+#                                   #
+#              VITS                 #
+#                                   #
+# ================================= #
+
+VITS_INFERENCE_ROOT_DIRECTORY="/vits"
+VITS_INFERENCE_SCRIPT="infer_ts_job.py"
+VITS_INFERENCE_MAYBE_PYTHON_INTERPRETER="python3"
+VITS_INFERENCE_MAYBE_DOCKER_IMAGE_SHA="03017896367b"
+
+# ================================= #
+#                                   #
+#          CACHE SETTINGS           #
+#                                   #
+# ================================= #
+
+# Huggingface caches data at ~/.cache/huggingface (/root/cache/huggingface),
+# which isn't the shared volume and will get blown away.
+# HF_HOME="/tmp/huggingface"
+HF_DATASETS_CACHE="/tmp/huggingface"
+
+# nltk also caches data.
+NLTK_DATA="/tmp/nltk_data"

crates/service/job/email_sender_job/config/email-sender-job.production.env

@@ -0,0 +1 @@
+# inference-job default .env configurations (these can be overridden with k8s)

crates/service/job/email_sender_job/src/http_server/endpoints/health_check_handler.rs

@@ -0,0 +1,109 @@
+use std::sync::Arc;
+
+use actix_web::{HttpRequest, HttpResponse, web};
+use actix_web::error::ResponseError;
+use actix_web::http::header::ContentType;
+use actix_web::http::StatusCode;
+use log::error;
+
+use actix_helpers::response_serializers::error_to_json_http_response::error_to_json_http_response;
+
+use crate::http_server::http_server_shared_state::HttpServerSharedState;
+
+// =============== Success Response ===============
+
+#[derive(Serialize)]
+pub struct HealthCheckResponse {
+  pub success: bool,
+  pub is_healthy: bool,
+
+  pub consecutive_failure_count: u64,
+  pub consecutive_success_count: u64,
+
+  pub total_failure_count: u64,
+  pub total_success_count: u64,
+
+  pub total_failure_ratio: f32,
+  pub total_success_ratio: f32,
+}
+
+
+// =============== Error Response ===============
+
+#[derive(Debug, Serialize)]
+pub enum HealthCheckError {
+  ServerError,
+}
+
+impl ResponseError for HealthCheckError {
+  fn status_code(&self) -> StatusCode {
+    match *self {
+      HealthCheckError::ServerError => StatusCode::INTERNAL_SERVER_ERROR,
+    }
+  }
+
+  fn error_response(&self) -> HttpResponse {
+    error_to_json_http_response(self)
+  }
+}
+
+// NB: Not using derive_more::Display since Clion doesn't understand it.
+impl std::fmt::Display for HealthCheckError {
+  fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+    write!(f, "{:?}", self)
+  }
+}
+
+// =============== Handler ===============
+
+pub async fn get_health_check_handler(
+  _http_request: HttpRequest,
+  server_state: web::Data<Arc<HttpServerSharedState>>
+) -> Result<HttpResponse, HealthCheckError> {
+  let job_stats = server_state.job_stats.get_status()
+      .map_err(|e| {
+        error!("Error serving health check status: {:?}", e);
+        HealthCheckError::ServerError
+      })?;
+
+  let total_tries = job_stats.total_failure_count.saturating_add(job_stats.total_success_count);
+
+  let total_success_ratio = if total_tries > 0 {
+    (job_stats.total_success_count as f32) / (total_tries as f32)
+  } else {
+    0.0
+  };
+
+  let total_failure_ratio = if total_tries > 0 {
+    1.0 - total_success_ratio
+  } else {
+    0.0
+  };
+
+  let is_healthy =
+      job_stats.consecutive_failure_count < server_state.consecutive_failure_unhealthy_threshold;
+
+  let response = HealthCheckResponse {
+    success: true,
+    is_healthy,
+    consecutive_failure_count: job_stats.consecutive_failure_count,
+    consecutive_success_count: job_stats.consecutive_success_count,
+    total_failure_count: job_stats.total_failure_count,
+    total_success_count: job_stats.total_success_count,
+    total_failure_ratio,
+    total_success_ratio,
+  };
+
+  let body = serde_json::to_string(&response)
+      .map_err(|e| HealthCheckError::ServerError)?;
+
+  if is_healthy {
+    Ok(HttpResponse::Ok()
+        .content_type(ContentType::json())
+        .body(body))
+  } else {
+    Ok(HttpResponse::InternalServerError()
+        .content_type(ContentType::json())
+        .body(body))
+  }
+}

crates/service/job/email_sender_job/src/http_server/endpoints/mod.rs

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

crates/service/job/email_sender_job/src/http_server/http_server_shared_state.rs

@@ -0,0 +1,7 @@
+use jobs_common::job_stats::JobStats;
+
+#[derive(Clone)]
+pub struct HttpServerSharedState {
+  pub job_stats: JobStats,
+  pub consecutive_failure_unhealthy_threshold: u64,
+}

crates/service/job/email_sender_job/src/http_server/mod.rs

@@ -0,0 +1,3 @@
+pub mod endpoints;
+pub mod http_server_shared_state;
+pub mod run_http_server;

crates/service/job/email_sender_job/src/http_server/run_http_server.rs

@@ -0,0 +1,72 @@
+use std::sync::Arc;
+
+use actix_web::{App, HttpResponse, HttpServer, web};
+use actix_web::dev::Server;
+use actix_web::middleware::Logger;
+use log::info;
+
+use bootstrap::bootstrap::ContainerEnvironment;
+use errors::AnyhowResult;
+use jobs_common::job_stats::JobStats;
+
+use crate::http_server::endpoints::health_check_handler::get_health_check_handler;
+use crate::http_server::http_server_shared_state::HttpServerSharedState;
+
+const DEFAULT_BIND_ADDRESS : &str = "0.0.0.0:12345";
+const DEFAULT_NUM_WORKERS : usize = 4;
+
+pub struct CreateServerArgs {
+  pub container_environment: ContainerEnvironment,
+  pub job_stats: JobStats,
+}
+
+pub fn run_http_server(args: CreateServerArgs) -> AnyhowResult<Server>
+{
+  // HTTP server args
+  let bind_address = easyenv::get_env_string_or_default("HTTP_BIND_ADDRESS", DEFAULT_BIND_ADDRESS);
+  let num_workers = easyenv::get_env_num("HTTP_NUM_WORKERS", DEFAULT_NUM_WORKERS)?;
+  let hostname = args.container_environment.hostname.clone();
+
+  let server_state = HttpServerSharedState {
+    job_stats: args.job_stats.clone(),
+    consecutive_failure_unhealthy_threshold:
+      easyenv::get_env_num("CONSECUTIVE_FAILURE_UNHEALTHY_THRESHOLD", 3)?,
+  };
+
+  let server_state_arc = web::Data::new(Arc::new(server_state));
+
+  info!("Starting HTTP service (just for k8s health checking).");
+
+  // NB: We shouldn't be logging much as the /_status endpoint is all we aim to expose.
+  let log_format = "[%{HOSTNAME}e] IP=[%{X-Forwarded-For}i] \"%r\" %s %b \"%{Referer}i\" \"%{User-Agent}i\" %T";
+
+  let handle = HttpServer::new(move || {
+    // NB: app_data being clone()'d below should all be safe (dependencies included)
+    App::new()
+        .app_data(server_state_arc.clone())
+        .wrap(Logger::new(&log_format)
+            .exclude("/_status")
+        )
+        //.wrap(middleware::Compress::default())
+        .service(
+          web::resource("/")
+              .route(web::get().to(|| HttpResponse::Ok()))
+              .route(web::head().to(|| HttpResponse::Ok()))
+        )
+        .service(
+          web::resource("/_status")
+              .route(web::get().to(get_health_check_handler))
+              .route(web::head().to(|| HttpResponse::Ok()))
+        )
+  })
+      .bind(bind_address)?
+      .workers(num_workers)
+      .run();
+
+  Ok(handle)
+}
+
+pub async fn launch_http_server(args: CreateServerArgs) -> AnyhowResult<()> {
+  run_http_server(args)?.await?;
+  Ok(())
+}

crates/service/job/email_sender_job/src/job/email_types/mod.rs

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

crates/service/job/email_sender_job/src/job/email_types/password_reset_email_sender.rs

@@ -0,0 +1,96 @@
+use anyhow::anyhow;
+
+use email_sender::letter_exports;
+use email_sender::letter_exports::Message;
+use mysql_queries::payloads::email_sender_jobs::email_sender_job_args::PolymorphicEmailSenderJobArgs;
+use mysql_queries::queries::email_sender_jobs::list_available_email_sender_jobs::AvailableEmailSenderJob;
+use server_environment::ServerEnvironment;
+
+use crate::job::job_loop::process_single_job_error::ProcessSingleJobError;
+use crate::job_dependencies::JobDependencies;
+
+pub async fn password_reset_email_sender(job: &AvailableEmailSenderJob, job_dependencies: &JobDependencies) -> Result<(), ProcessSingleJobError> {
+
+  let maybe_inner_args = job.maybe_email_args
+      .as_ref()
+      .and_then(|args| args.args.as_ref());
+
+  let maybe_secret_key = match maybe_inner_args {
+    Some(PolymorphicEmailSenderJobArgs::Pr(reset_args)) => {
+      reset_args.password_reset_secret_key.clone()
+    }
+    _ => {
+      log::error!("Incorrect job args for password reset!");
+      return Err(ProcessSingleJobError::InvalidJob(anyhow!("Incorrect job args for password reset!")));
+    },
+  };
+
+  let secret_key = match maybe_secret_key {
+    Some(key) => key,
+    None => {
+      return Err(ProcessSingleJobError::InvalidJob(anyhow!("No secret key for password reset!")));
+    }
+  };
+
+  match &job.maybe_email_args {
+    Some(args) => {
+      match args.args {
+        None => {}
+        Some(_) => {}
+      }
+
+    }
+    None => {}
+  }
+
+  let from_address = "Support <support@storyteller.ai>"
+      .parse()
+      .map_err(|err| {
+        log::error!("Error parsing from address: {err}");
+        ProcessSingleJobError::InvalidJob(anyhow!("Error parsing from address: {err}"))
+      })?;
+
+  let to_address = job.destination_email_address
+      .parse()
+      .map_err(|err| {
+        log::error!("Error parsing to address: {err}");
+        ProcessSingleJobError::InvalidJob(anyhow!("Error parsing to address: {err}"))
+      })?;
+
+  // TODO(bt,2023-11-12): Environmentally configure, allow overrides.
+  let link = match job_dependencies.server_environment {
+    ServerEnvironment::Development => format!("http://dev.fakeyou.com:7000/password-reset/verify?token={secret_key}"),
+    ServerEnvironment::Production => format!("https://fakeyou.com/password-reset/verify?token={secret_key}"),
+  };
+
+  let message = format!(r#"
+      <a href="{link}">Click here to reset your password!</a>
+      <br />
+      <br />
+      If you can't click the link, here's the secret reset code: {secret_key}
+      <br />
+      <br />
+      Thank You,
+      <br />
+      <br />
+      Storyteller.ai (FakeYou) Team
+    "#);
+
+  let email = Message::builder()
+      .from(from_address)
+      .to(to_address)
+      .subject("FakeYou Password Reset")
+      .header(letter_exports::ContentType::TEXT_HTML)
+      .body(message)
+      .map_err(|err| {
+        log::error!("Error constructing email: {err}");
+        ProcessSingleJobError::InvalidJob(anyhow!("Error constructing email: {err}"))
+      })?;
+
+  job_dependencies.email_sender.send_message(&email).map_err(|err| {
+    log::error!("Error sending email: {err}");
+    ProcessSingleJobError::Other(anyhow!("Error sending email: {err}"))
+  })?;
+
+  Ok(())
+}

crates/service/job/email_sender_job/src/job/job_loop/main_loop.rs

@@ -0,0 +1,212 @@
+use std::time::Duration;
+
+use log::{error, info, warn};
+
+use errors::AnyhowResult;
+use filesys::file_exists::file_exists;
+use jobs_common::noop_logger::NoOpLogger;
+use mysql_queries::queries::email_sender_jobs::list_available_email_sender_jobs::{AvailableEmailSenderJob, list_available_email_sender_jobs, ListAvailableEmailSenderJobArgs};
+use mysql_queries::queries::email_sender_jobs::mark_email_sender_job_completely_failed::mark_email_sender_job_completely_failed;
+use mysql_queries::queries::email_sender_jobs::mark_email_sender_job_failure::mark_email_sender_job_failure;
+
+use crate::job::job_loop::process_single_job::process_single_job;
+use crate::job::job_loop::process_single_job_error::ProcessSingleJobError;
+use crate::job::job_loop::process_single_job_success_case::ProcessSingleJobSuccessCase;
+use crate::job_dependencies::JobDependencies;
+
+// Job runner timeouts (guards MySQL)
+const START_TIMEOUT_MILLIS : u64 = 500;
+const INCREASE_TIMEOUT_MILLIS : u64 = 1000;
+
+/// Pause file millis
+const PAUSE_FILE_EXISTS_WAIT_MILLIS : u64 = 1000 * 30;
+
+pub async fn main_loop(job_dependencies: JobDependencies) {
+  let mut noop_logger = NoOpLogger::new(job_dependencies.no_op_logger_millis as i64);
+
+  let mut error_timeout_millis = START_TIMEOUT_MILLIS;
+  let mut sort_by_priority = true;
+  let mut sort_by_priority_count = 0;
+
+  while !job_dependencies.application_shutdown.get() {
+    if let Some(pause_file) = job_dependencies.fs.maybe_pause_file.as_deref() {
+      while file_exists(pause_file) {
+        warn!("Pause file exists. Pausing until deleted: {:?}", pause_file);
+        std::thread::sleep(Duration::from_millis(PAUSE_FILE_EXISTS_WAIT_MILLIS));
+      }
+    }
+
+    // Don't completely starve low-priority jobs
+    if sort_by_priority_count >= job_dependencies.low_priority_starvation_prevention_every_nth {
+      sort_by_priority_count = 0;
+      sort_by_priority = false;
+    }
+
+    let maybe_available_jobs = list_available_email_sender_jobs(ListAvailableEmailSenderJobArgs {
+      num_records: job_dependencies.job_batch_size,
+      is_debug_worker: false, // TODO
+      sort_by_priority,
+      mysql_pool: &job_dependencies.mysql_pool,
+    }).await;
+
+    sort_by_priority = true;
+    sort_by_priority_count += 1;
+
+    let jobs = match maybe_available_jobs {
+      Ok(jobs) => jobs,
+      Err(e) => {
+        error!("Error querying jobs: {:?}", e);
+        std::thread::sleep(Duration::from_millis(error_timeout_millis));
+        error_timeout_millis += INCREASE_TIMEOUT_MILLIS;
+        continue;
+      }
+    };
+
+    if jobs.is_empty() {
+      noop_logger.log_message_after_awhile("No jobs picked up from database!");
+
+      std::thread::sleep(Duration::from_millis(job_dependencies.job_batch_wait_millis));
+      error_timeout_millis = START_TIMEOUT_MILLIS; // reset
+      continue;
+    }
+
+    info!("Queried {} jobs from database", jobs.len());
+
+    let batch_result = process_job_batch(&job_dependencies, jobs).await;
+
+    match batch_result {
+      Ok(_) => {},
+      Err(e) => {
+        error!("Error processing jobs: {:?}", e);
+        std::thread::sleep(Duration::from_millis(error_timeout_millis));
+        error_timeout_millis += INCREASE_TIMEOUT_MILLIS;
+        continue;
+      }
+    }
+
+    error_timeout_millis = START_TIMEOUT_MILLIS; // reset
+
+    std::thread::sleep(Duration::from_millis(job_dependencies.job_batch_wait_millis));
+  }
+
+  warn!("Job runner main loop is shut down.");
+}
+
+// TODO: A common interface/trait for each submodule (tts, webvc) to declare how to determine if the job is "ready".
+//  This probably returns a struct or enum with some measure of how many GB need to be downloaded.
+
+async fn process_job_batch(job_dependencies: &JobDependencies, jobs: Vec<AvailableEmailSenderJob>) -> AnyhowResult<()> {
+  for job in jobs.into_iter() {
+    let result = process_single_job(job_dependencies, &job).await;
+    match result {
+      Ok(success_case) => {
+        info!("Job loop iteration \"success\": {:?}", success_case);
+
+        let increment_success_count = match success_case {
+          ProcessSingleJobSuccessCase::JobCompleted => true,
+          ProcessSingleJobSuccessCase::JobSkippedForRoutingTagMismatch => false,
+          ProcessSingleJobSuccessCase::LockNotObtained => false,
+        };
+
+        if increment_success_count {
+          let stats = job_dependencies.job_stats.increment_success_count().ok();
+          warn!("Success stats: {:?}", stats);
+        }
+      },
+      Err(err) => {
+        warn!("Failure to process job: {:?}", err);
+        let _r = handle_error(&job_dependencies, &job, err).await?;
+      }
+    }
+  }
+
+  Ok(())
+}
+
+#[derive(Eq,PartialEq)]
+enum JobFailureClass {
+  // Jobs that can be retried
+  TransientFailure,
+  // Jobs that cannot be retried and must be marked dead
+  PermanentFailure,
+}
+
+#[derive(Eq,PartialEq)]
+enum ContainerHealth {
+  // No impact to container health
+  Ignore,
+  // Increment the container health failure counter
+  IncrementContainerFailCount,
+}
+
+async fn handle_error(job_dependencies: &&JobDependencies, job: &AvailableEmailSenderJob, error: ProcessSingleJobError) -> AnyhowResult<()> {
+  let (
+    job_failure_class,
+    container_health_report,
+    internal_failure_reason,
+  ) = match error {
+    // Permanent failures
+    ProcessSingleJobError::KeepAliveElapsed =>
+      (
+        JobFailureClass::PermanentFailure,
+        ContainerHealth::Ignore,
+        "keepalive elapsed".to_string(),
+      ),
+    ProcessSingleJobError::InvalidJob(ref err) =>
+      (
+        JobFailureClass::PermanentFailure,
+        ContainerHealth::Ignore,
+        format!("InvalidJob: {:?}", err),
+      ),
+    ProcessSingleJobError::NotYetImplemented =>
+      (
+        JobFailureClass::PermanentFailure,
+        ContainerHealth::Ignore,
+        "not yet implemented".to_string(),
+      ),
+
+    // Non-permanent failures
+    ProcessSingleJobError::Other(ref err) =>
+      (
+        JobFailureClass::TransientFailure,
+        ContainerHealth::IncrementContainerFailCount,
+        format!("OtherErr: {:?}", err),
+      ),
+  };
+
+  if container_health_report == ContainerHealth::IncrementContainerFailCount {
+    // NB: We only increment the fail count for events that may indicate the job server is stuck.
+    let stats = job_dependencies.job_stats.increment_failure_count().ok();
+    warn!("Failure stats: {:?}", stats);
+  }
+
+  match job_failure_class {
+    JobFailureClass::PermanentFailure => {
+      let _r = mark_email_sender_job_completely_failed(
+        &job_dependencies.mysql_pool,
+        &job,
+        Some(&internal_failure_reason),
+      ).await;
+    }
+    JobFailureClass::TransientFailure => {
+      let _r = mark_email_sender_job_failure(
+        &job_dependencies.mysql_pool,
+        &job,
+        &internal_failure_reason,
+        job_dependencies.job_max_attempts
+      ).await;
+    }
+  }
+
+  match error {
+    // Post failure handling
+    //   (none)
+    // No-op
+    ProcessSingleJobError::Other(_) => {}
+    ProcessSingleJobError::InvalidJob(_) => {}
+    ProcessSingleJobError::KeepAliveElapsed => {}
+    ProcessSingleJobError::NotYetImplemented => {}
+  }
+
+  Ok(())
+}

crates/service/job/email_sender_job/src/job/job_loop/mod.rs

@@ -0,0 +1,4 @@
+pub mod main_loop;
+pub mod process_single_job;
+pub mod process_single_job_error;
+pub mod process_single_job_success_case;

crates/service/job/email_sender_job/src/job/job_loop/process_single_job.rs

@@ -0,0 +1,105 @@
+use std::time::Instant;
+
+use anyhow::anyhow;
+use log::{info, warn};
+
+use enums::by_table::email_sender_jobs::email_category::EmailCategory;
+use mysql_queries::queries::email_sender_jobs::list_available_email_sender_jobs::AvailableEmailSenderJob;
+use mysql_queries::queries::email_sender_jobs::mark_email_sender_job_pending_and_grab_lock::mark_email_sender_job_pending_and_grab_lock;
+use mysql_queries::queries::email_sender_jobs::mark_email_sender_job_successfully_done::mark_email_sender_job_successfully_done;
+
+use crate::job::email_types::password_reset_email_sender::password_reset_email_sender;
+use crate::job::job_loop::process_single_job_error::ProcessSingleJobError;
+use crate::job::job_loop::process_single_job_success_case::ProcessSingleJobSuccessCase;
+use crate::job_dependencies::JobDependencies;
+
+pub async fn process_single_job(
+  job_dependencies: &JobDependencies,
+  job: &AvailableEmailSenderJob,
+) -> Result<ProcessSingleJobSuccessCase, ProcessSingleJobError> {
+  let mut force_execution = false;
+
+  // Some jobs have "routing tags". These ensure that jobs only execute on certain hosts.
+  // This is typically for debugging or development.
+  if let Some(routing_tag) = job.maybe_routing_tag.as_deref() {
+    let routing_tag = routing_tag.to_lowercase();
+    let hostname = job_dependencies.container.hostname.to_ascii_lowercase();
+
+    if hostname.starts_with(&routing_tag) {
+      info!("Job has routing tag ({}) for execution on this host ({})", routing_tag, hostname);
+      force_execution = true;
+    } else {
+      info!("Job routing tag ({}) doesn't match hostname ({}); skipping...", routing_tag, hostname);
+      return Ok(ProcessSingleJobSuccessCase::JobSkippedForRoutingTagMismatch);
+    }
+  }
+
+  // ==================== ATTEMPT TO GRAB JOB LOCK ==================== //
+
+  let lock_acquired = mark_email_sender_job_pending_and_grab_lock(
+    &job_dependencies.mysql_pool,
+    job.id,
+    &job_dependencies.container_db,
+  ).await
+      .map_err(|err| ProcessSingleJobError::Other(anyhow!("database error: {:?}", err)))?;
+
+  if !lock_acquired {
+    warn!("Could not acquire job lock for: {}", &job.id.0);
+    return Ok(ProcessSingleJobSuccessCase::LockNotObtained)
+  }
+
+  process_single_job_wrap_with_logs(job_dependencies, job).await
+}
+
+async fn process_single_job_wrap_with_logs(
+  job_dependencies: &JobDependencies,
+  job: &AvailableEmailSenderJob,
+) -> Result<ProcessSingleJobSuccessCase, ProcessSingleJobError> {
+
+  println!("\n  ----------------------------------------- JOB START -----------------------------------------  \n");
+
+  info!("Beginning work on job ({}): {:?}", job.id.0, job.token);
+
+  let result = do_process_single_job(job_dependencies, job).await;
+
+  println!("\n  ----------------------------------------- JOB END -----------------------------------------  \n");
+
+  result
+}
+
+async fn do_process_single_job(
+  job_dependencies: &JobDependencies,
+  job: &AvailableEmailSenderJob,
+) -> Result<ProcessSingleJobSuccessCase, ProcessSingleJobError> {
+
+  let job_start_time = Instant::now();
+
+  // ==================== HANDLE DIFFERENT INFERENCE TYPES ==================== //
+
+  match job.email_category {
+    EmailCategory::Welcome => {
+      return Err(ProcessSingleJobError::NotYetImplemented);
+    }
+    EmailCategory::PasswordReset => {
+      let _r = password_reset_email_sender(job, job_dependencies).await?;
+    }
+  };
+
+  // =====================================================
+
+  let job_duration = Instant::now().duration_since(job_start_time);
+
+  info!("Job took duration to complete: {:?}", &job_duration);
+
+  info!("Marking job complete...");
+
+  mark_email_sender_job_successfully_done(
+    &job_dependencies.mysql_pool,
+    job,
+  ).await
+      .map_err(|err| ProcessSingleJobError::Other(anyhow!("database error: {:?}", err)))?;
+
+  info!("Job done: {} : {:?}", job.id.0, job.token);
+
+  Ok(ProcessSingleJobSuccessCase::JobCompleted)
+}

crates/service/job/email_sender_job/src/job/job_loop/process_single_job_error.rs

@@ -0,0 +1,18 @@
+/// Error from processing a single job
+#[derive(Debug)]
+pub enum ProcessSingleJobError {
+  /// The job is invalid (bad state, etc.)
+  InvalidJob(anyhow::Error),
+  /// The job's keepalive signal has elapsed, so the job needs to be killed.
+  /// The keepalive signal is typically for non-premium users only and it
+  /// makes sure they're still on the website. It kills jobs they leave
+  /// running when they bounce so that we can free up their capacity.
+  KeepAliveElapsed,
+  /// This is any other kind of error.
+  /// It might be important, we just haven't special cased it yet.
+  Other(anyhow::Error),
+ 
+  /// We hit a feature or a path for this job that has not yet been implemented.
+  /// Permanently fail the job.
+  NotYetImplemented
+}

crates/service/job/email_sender_job/src/job/job_loop/process_single_job_success_case.rs

@@ -0,0 +1,14 @@
+
+#[derive(Copy, Clone, Debug)]
+pub enum ProcessSingleJobSuccessCase {
+  /// Job was successfully completed.
+  JobCompleted,
+
+  /// The job has a routing tag that doesn't match this container.
+  /// This job will never execute on this container.
+  JobSkippedForRoutingTagMismatch,
+
+  /// The lock for the job could not be obtained. Another worker might have it.
+  /// If that workload fails, we could conceivably pick it up again in the future.
+  LockNotObtained,
+}

crates/service/job/email_sender_job/src/job/mod.rs

@@ -0,0 +1,2 @@
+pub mod email_types;
+pub mod job_loop;

crates/service/job/email_sender_job/src/job_dependencies.rs

@@ -0,0 +1,78 @@
+use std::path::PathBuf;
+
+use r2d2_redis::r2d2;
+use r2d2_redis::RedisConnectionManager;
+use sqlx::MySqlPool;
+
+use bootstrap::bootstrap::ContainerEnvironment;
+use concurrency::relaxed_atomic_bool::RelaxedAtomicBool;
+use email_sender::smtp_email_sender::SmtpEmailSender;
+use jobs_common::job_progress_reporter::job_progress_reporter::JobProgressReporterBuilder;
+use jobs_common::job_stats::JobStats;
+use mysql_queries::common_inputs::container_environment_arg::ContainerEnvironmentArg;
+use server_environment::ServerEnvironment;
+
+pub struct JobDependencies {
+  /// Filesystem info and utils
+  pub fs: FileSystemDetails,
+
+  pub mysql_pool: MySqlPool,
+
+  // TODO(2023-01-11): We don't always connect to a Redis
+  //  Typically this is for job status reporting, but we might also report on when users leave the
+  //  site to proactively kill their inference jobs and save on worker quota.
+  //  On local dev we probably don't care about Redis at all, and on on-prem workers, we cannot
+  //  connect to production Redis easily (requires lots of setup - ghosttunnel or something + IP rules)
+  pub maybe_redis_pool: Option<r2d2::Pool<RedisConnectionManager>>,
+
+  pub email_sender: SmtpEmailSender,
+
+  pub server_environment: ServerEnvironment,
+
+  pub maybe_keepalive_redis_pool: Option<r2d2::Pool<RedisConnectionManager>>,
+
+  pub job_progress_reporter: Box<dyn JobProgressReporterBuilder>,
+
+  pub job_stats: JobStats,
+
+  pub worker_details: JobWorkerDetails,
+
+  // Sleep between batches
+  pub job_batch_wait_millis: u64,
+
+  // Max job attempts before failure.
+  pub job_max_attempts: u16,
+
+  // Number of jobs to dequeue at once.
+  pub job_batch_size: u32,
+
+  // How long to wait between log lines
+  pub no_op_logger_millis: u64,
+
+  // Typically we'll sort jobs by priority. Occasionally we introduce a chance for low
+  // priority jobs to run in the order they were enqueued.
+  // If this is set to "0", we no longer consider priority
+  pub low_priority_starvation_prevention_every_nth: usize,
+
+  pub container: ContainerEnvironment,
+  pub container_db: ContainerEnvironmentArg, // Same info, but for database.
+
+  // The application can be shut down from another thread.
+  // Checking this will determine if the application needs to exit (true = exit).
+  pub application_shutdown: RelaxedAtomicBool,
+}
+
+pub struct FileSystemDetails {
+  /// If the pause file is set and exists on the filesystem,
+  /// the job will pause until the file stops existing.
+  /// Good for live debugging of production k8s clusters without
+  /// redeploying.
+  pub maybe_pause_file: Option<PathBuf>,
+}
+
+pub struct JobWorkerDetails {
+  // Debug workers only process special debug requests. They're silent otherwise.
+  // Non-debug workers ignore debug requests. This is so we can deploy special code
+  // to debug nodes (typically just one, perhaps even ephemerally).
+  pub is_debug_worker: bool,
+}

crates/service/job/email_sender_job/src/main.rs

@@ -0,0 +1,187 @@
+// Never allow these
+#![forbid(private_in_public)]
+#![forbid(unused_must_use)] // NB: It's unsafe to not close/check some things
+
+// Okay to toggle
+//#![forbid(unreachable_patterns)]
+//#![forbid(unused_imports)]
+//#![forbid(unused_mut)]
+//#![forbid(unused_variables)]
+
+// Always allow
+#![allow(dead_code)]
+#![allow(non_snake_case)]
+
+// Strict AF
+//#![forbid(warnings)]
+
+#[macro_use] extern crate serde_derive;
+
+use anyhow::anyhow;
+use log::{info, warn};
+use r2d2_redis::r2d2;
+use r2d2_redis::RedisConnectionManager;
+use sqlx::mysql::MySqlPoolOptions;
+
+use bootstrap::bootstrap::{bootstrap, BootstrapArgs};
+use concurrency::relaxed_atomic_bool::RelaxedAtomicBool;
+use config::common_env::CommonEnv;
+use config::shared_constants::DEFAULT_MYSQL_CONNECTION_STRING;
+use config::shared_constants::DEFAULT_RUST_LOG;
+use container_common::anyhow_result::AnyhowResult;
+use email_sender::smtp_email_sender::SmtpEmailSender;
+use jobs_common::job_progress_reporter::job_progress_reporter::JobProgressReporterBuilder;
+use jobs_common::job_progress_reporter::noop_job_progress_reporter::NoOpJobProgressReporterBuilder;
+use jobs_common::job_progress_reporter::redis_job_progress_reporter::RedisJobProgressReporterBuilder;
+use jobs_common::job_stats::JobStats;
+use mysql_queries::common_inputs::container_environment_arg::ContainerEnvironmentArg;
+use server_environment::ServerEnvironment;
+
+use crate::http_server::run_http_server::CreateServerArgs;
+use crate::http_server::run_http_server::launch_http_server;
+use crate::job::job_loop::main_loop::main_loop;
+use crate::job_dependencies::{FileSystemDetails, JobDependencies, JobWorkerDetails};
+
+pub mod http_server;
+pub mod job;
+pub mod job_dependencies;
+
+// Buckets (shared config)
+const ENV_ACCESS_KEY : &str = "ACCESS_KEY";
+const ENV_SECRET_KEY : &str = "SECRET_KEY";
+const ENV_REGION_NAME : &str = "REGION_NAME";
+
+// Bucket names
+const ENV_PRIVATE_BUCKET_NAME : &str = "PRIVATE_BUCKET_NAME";
+const ENV_PUBLIC_BUCKET_NAME : &str = "PUBLIC_BUCKET_NAME";
+
+// HTTP sidecar
+const ENV_TTS_INFERENCE_SIDECAR_HOSTNAME: &str = "TTS_INFERENCE_SIDECAR_HOSTNAME";
+
+#[actix_web::main]
+async fn main() -> AnyhowResult<()> {
+
+  let container_environment = bootstrap(BootstrapArgs {
+    app_name: "email-sender-job",
+    default_logging_override: Some(DEFAULT_RUST_LOG),
+    config_search_directories: &[".", "./config", "crates/service/job/email_sender_job/config"],
+  })?;
+
+  info!("Hostname: {}", &container_environment.hostname);
+
+  // NB: These are non-standard env vars we're injecting ourselves.
+  let k8s_node_name = easyenv::get_env_string_optional("K8S_NODE_NAME");
+  let k8s_pod_name = easyenv::get_env_string_optional("K8S_POD_NAME");
+
+  let db_connection_string =
+      easyenv::get_env_string_or_default(
+        "MYSQL_URL",
+        DEFAULT_MYSQL_CONNECTION_STRING);
+
+  info!("Connecting to database...");
+
+  let mysql_pool = MySqlPoolOptions::new()
+      .max_connections(2)
+      .connect(&db_connection_string)
+      .await?;
+
+  let common_env = CommonEnv::read_from_env()?;
+
+  // Set to "0" to always treat low priority the same as high priority
+  let low_priority_starvation_prevention_every_nth= easyenv::get_env_num(
+    "LOW_PRIORITY_STARVATION_PREVENTION_EVERY_NTH", 3)?;
+
+  let server_environment = ServerEnvironment::from_str(&easyenv::get_env_string_required("SERVER_ENVIRONMENT")?)
+      .ok_or(anyhow!("invalid server environment"))?;
+
+  // TODO(bt,2023-11-11): Password and account details should be a secret, but gotta go fast.
+  let email_sender = SmtpEmailSender::new(
+    "smtp.gmail.com",
+    "noreply@storyteller.ai".to_string(),
+    "FakeYouHanashi1".to_string())?;
+
+  let is_debug_worker = easyenv::get_env_bool_or_default("IS_DEBUG_WORKER", false);
+
+  info!("Is debug worker? {}", is_debug_worker);
+
+  // Optionally report job progress to the user via Redis (for now)
+  // We want to turn this off in the on-premises workers since we're not tunneling to the production Redis.
+  let job_progress_reporter : Box<dyn JobProgressReporterBuilder>;
+
+  job_progress_reporter = match easyenv::get_env_string_optional("REDIS_FOR_JOB_PROGRESS") {
+    None => {
+      warn!("Redis for job progress status reports is DISABLED! Users will not see in-flight details of inference progress.");
+      Box::new(NoOpJobProgressReporterBuilder {})
+    },
+    Some(redis_connection_string) => {
+      info!("Connecting to Redis to use for reporting job progress... {}", redis_connection_string);
+      let redis_manager = RedisConnectionManager::new(redis_connection_string)?;
+      let redis_pool = r2d2::Pool::builder().build(redis_manager)?;
+
+      Box::new(RedisJobProgressReporterBuilder::from_redis_pool(redis_pool))
+    }
+  };
+
+  let maybe_keepalive_redis_pool =
+      match easyenv::get_env_string_optional("REDIS_FOR_KEEPALIVE_URL") {
+        None => None,
+        Some(redis_url) => {
+          let redis_manager = RedisConnectionManager::new(redis_url)?;
+          let redis_pool = r2d2::Pool::builder().build(redis_manager)?;
+          Some(redis_pool)
+        }
+      };
+
+  // NB: Threading eats the Ctrl-C signal, so we're going to send application shutdown across
+  // threads with an atomic bool.
+  let application_shutdown = RelaxedAtomicBool::new(false);
+
+  let job_stats = JobStats::new();
+
+  let create_server_args = CreateServerArgs {
+    container_environment: container_environment.clone(),
+    job_stats: job_stats.clone(),
+  };
+
+  let job_dependencies = JobDependencies {
+    fs: FileSystemDetails {
+      maybe_pause_file: easyenv::get_env_pathbuf_optional("PAUSE_FILE"),
+    },
+    mysql_pool,
+    maybe_redis_pool: None, // TODO(bt, 2023-01-11): See note in JobDependencies
+    maybe_keepalive_redis_pool,
+    job_progress_reporter,
+    job_stats,
+    worker_details: JobWorkerDetails {
+      is_debug_worker,
+    },
+    server_environment,
+    email_sender,
+    job_batch_wait_millis: common_env.job_batch_wait_millis,
+    job_max_attempts: common_env.job_max_attempts as u16,
+    job_batch_size: common_env.job_batch_size,
+    no_op_logger_millis: common_env.no_op_logger_millis,
+    low_priority_starvation_prevention_every_nth,
+    container: container_environment.clone(),
+    container_db: ContainerEnvironmentArg {
+      hostname: container_environment.hostname,
+      cluster_name: container_environment.cluster_name,
+    },
+    application_shutdown: application_shutdown.clone(),
+  };
+
+  std::thread::spawn(move || {
+    let actix_runtime = actix_web::rt::System::new();
+    let http_server_handle = launch_http_server(create_server_args);
+
+    actix_runtime.block_on(http_server_handle)
+        .expect("HTTP server should not exit.");
+
+    warn!("Server thread is shut down.");
+    application_shutdown.set(true);
+  });
+
+  main_loop(job_dependencies).await;
+
+  Ok(())
+}

crates/service/plugins/users/src/endpoints/password_reset_request_handler.rs

@@ -7,13 +7,15 @@ use serde::Deserialize;
 use sqlx::MySqlPool;
 use strum_macros::Display;
 
+use container_common::token::random_uuid::generate_random_uuid;
 use crockford::crockford_entropy_lower;
-use email_sender::letter_exports;
-use email_sender::letter_exports::Message;
 use email_sender::smtp_email_sender::SmtpEmailSender;
-use errors::AnyhowResult;
+use enums::by_table::email_sender_jobs::email_category::EmailCategory;
 use http_server_common::request::get_request_ip::get_request_ip;
 use http_server_common::response::serialize_as_json_error::serialize_as_json_error;
+use mysql_queries::payloads::email_sender_jobs::email_sender_job_args::{EmailSenderJobArgs, PolymorphicEmailSenderJobArgs};
+use mysql_queries::payloads::email_sender_jobs::subtypes::email_job_password_reset_args::EmailJobPasswordResetArgs;
+use mysql_queries::queries::email_sender_jobs::insert_email_sender_job::{insert_email_sender_job, InsertEmailSenderJobArgs};
 use mysql_queries::queries::users::user::lookup_user_for_login_by_email::lookup_user_for_login_by_email;
 use mysql_queries::queries::users::user::lookup_user_for_login_by_username::lookup_user_for_login_by_username;
 use mysql_queries::queries::users::user_password_resets::create_password_reset_request::create_password_reset;
@@ -79,8 +81,8 @@ pub async fn password_reset_request_handler(
     http_request: HttpRequest,
     request: web::Json<PasswordResetRequestedRequest>,
     mysql_pool: web::Data<MySqlPool>,
-    server_environment: web::Data<ServerEnvironment>,
-    sender: web::Data<SmtpEmailSender>,
+    _server_environment: web::Data<ServerEnvironment>,
+    _sender: web::Data<SmtpEmailSender>,
 ) -> Result<HttpResponse, PasswordResetRequestedErrorResponse> {
 
     let username_or_email = request.username_or_email.trim();
@@ -106,69 +108,54 @@ pub async fn password_reset_request_handler(
 
     let secret_key = crockford_entropy_lower(32);
 
-    //TODO: Handle banned users, they shouldn't be able to do this
+    // TODO(bt,2023-11-15): Handle banned users, they shouldn't be able to do this
+
+    // TODO(bt,2023-11-15): AVT cookie
+    //let maybe_avt_token = server_state.avt_cookie_manager.get_avt_token_from_request(&http_request);
 
     let ip_address = get_request_ip(&http_request);
 
-    create_password_reset(&mysql_pool, &user, ip_address, secret_key.clone()).await
+    create_password_reset(&mysql_pool, &user, &ip_address, secret_key.clone()).await
         .map_err(|err| {
             log::error!("Error creating password reset: {err}");
             PasswordResetRequestedRequestError::Internal
         })?;
 
-    let from_address = "Support <support@storyteller.ai>"
-        .parse()
+    let uuid_idempotency_token = generate_random_uuid();
+
+    // TODO(bt,2023-11-15): i18n
+    let ietf_language_tag = "en";
+    let ietf_primary_language_subtag = "en";
+
+    insert_email_sender_job(InsertEmailSenderJobArgs {
+        uuid_idempotency_token: &uuid_idempotency_token,
+        destination_email_address: &user.email_address,
+        maybe_destination_user_token: Some(&user.token),
+        email_category: EmailCategory::PasswordReset,
+        maybe_email_args: Some(EmailSenderJobArgs {
+            args: Some(PolymorphicEmailSenderJobArgs::Pr(EmailJobPasswordResetArgs {
+                password_reset_secret_key: Some(secret_key),
+            })),
+        }),
+        ietf_language_tag,
+        ietf_primary_language_subtag,
+        maybe_creator_user_token: None,
+        maybe_avt_token: None,
+        creator_ip_address: &ip_address,
+        priority_level: 1,
+        is_debug_request: false,
+        maybe_routing_tag: None,
+        mysql_pool: &mysql_pool,
+    })
+        .await
         .map_err(|err| {
-            log::error!("Error parsing from address: {err}");
+            log::error!("Error inserting email job: {err}");
             PasswordResetRequestedRequestError::Internal
         })?;
 
-    let to_address = user.email_address
-        .parse()
-        .map_err(|err| {
-            log::error!("Error parsing to address: {err}");
-            PasswordResetRequestedRequestError::Internal
-        })?;
-
-    // TODO(bt,2023-11-12): Environmentally configure, allow overrides.
-    let link = match **server_environment {
-        ServerEnvironment::Development => format!("http://dev.fakeyou.com:7000/password-reset/verify?token={secret_key}"),
-        ServerEnvironment::Production => format!("https://fakeyou.com/password-reset/verify?token={secret_key}"),
-    };
-
-    let message = format!(r#"
-      <a href="{link}">Click here to reset your password!</a>
-      <br />
-      <br />
-      If you can't click the link, here's the secret reset code: {secret_key}
-      <br />
-      <br />
-      Thank You,
-      <br />
-      <br />
-      Storyteller.ai (FakeYou) Team
-    "#);
-
-    let email = Message::builder()
-        .from(from_address)
-        .to(to_address)
-        .subject("FakeYou Password Reset")
-        .header(letter_exports::ContentType::TEXT_HTML)
-        .body(message)
-        .map_err(|err| {
-            log::error!("Error constructing email: {err}");
-            PasswordResetRequestedRequestError::Internal
-        })?;
-
-    sender.send_message(&email).map_err(|err| {
-        log::error!("Error sending email: {err}");
-        PasswordResetRequestedRequestError::Internal
-    })?;
-
     success_response()
 }
 
-
 fn success_response() -> Result<HttpResponse, PasswordResetRequestedErrorResponse> {
     let response = PasswordResetRequestedResponse {
         success: true,

service_cpu.Dockerfile

@@ -123,7 +123,13 @@ RUN SQLX_OFFLINE=true \
   LD_LIBRARY_PATH=/usr/lib:${LD_LIBRARY_PATH} \
   $HOME/.cargo/bin/cargo build \
   --release \
-  --bin tts-download-job
+  --bin tts-download-job \
+
+RUN SQLX_OFFLINE=true \
+  LD_LIBRARY_PATH=/usr/lib:${LD_LIBRARY_PATH} \
+  $HOME/.cargo/bin/cargo build \
+  --release \
+  --bin email-sender-job
 
 RUN SQLX_OFFLINE=true \
   LD_LIBRARY_PATH=/usr/lib:${LD_LIBRARY_PATH} \
@@ -185,6 +191,7 @@ RUN echo -n ${GIT_SHA} > GIT_SHA
 COPY --from=builder /tmp/target/release/storyteller-web /
 COPY --from=builder /tmp/target/release/dummy-service /
 COPY --from=builder /tmp/target/release/analytics-job /
+COPY --from=builder /tmp/target/release/email-sender-job  /
 
 # Legacy apps:
 COPY --from=builder /tmp/target/release/tts-download-job /