From c7c7c5b7984f106e716623aea9aa5a29128413b8 Mon Sep 17 00:00:00 2001 From: Brandon Thomas Date: Wed, 15 Nov 2023 04:06:19 -0500 Subject: [PATCH] email-job --- Cargo.lock | 52 ++++ Cargo.toml | 1 + .../down.sql | 5 + .../up.sql | 160 ++++++++++ crates/_types/enums/Cargo.toml | 4 + .../email_sender_jobs/email_category.rs | 115 ++++++++ .../src/by_table/email_sender_jobs/mod.rs | 1 + crates/_types/enums/src/by_table/mod.rs | 1 + .../enums/src/common/job_status_plus.rs | 4 +- crates/_types/tokens/src/prefixes.rs | 5 +- .../src/tokens/email_sender_job_token.rs | 14 + crates/_types/tokens/src/tokens/mod.rs | 1 + .../email_sender_job_args.rs | 39 +++ .../src/payloads/email_sender_jobs/mod.rs | 2 + .../subtypes/email_job_password_reset_args.rs | 29 ++ .../subtypes/email_job_welcome_args.rs | 5 + .../email_sender_jobs/subtypes/mod.rs | 2 + crates/lib/mysql_queries/src/payloads/mod.rs | 1 + .../src/queries/email_sender_jobs/_keys.rs | 6 + .../insert_email_sender_job.rs | 112 +++++++ .../list_available_email_sender_jobs.rs | 277 ++++++++++++++++++ ...mark_email_sender_job_completely_failed.rs | 41 +++ .../mark_email_sender_job_failure.rs | 49 ++++ ..._email_sender_job_pending_and_grab_lock.rs | 118 ++++++++ ...mark_email_sender_job_successfully_done.rs | 32 ++ .../src/queries/email_sender_jobs/mod.rs | 7 + crates/lib/mysql_queries/src/queries/mod.rs | 3 +- .../create_password_reset_request.rs | 2 +- .../service/job/email_sender_job/Cargo.toml | 88 ++++++ .../config/email-sender-job.common.env | 1 + .../config/email-sender-job.development.env | 109 +++++++ .../config/email-sender-job.production.env | 1 + .../endpoints/health_check_handler.rs | 109 +++++++ .../src/http_server/endpoints/mod.rs | 1 + .../http_server/http_server_shared_state.rs | 7 + .../email_sender_job/src/http_server/mod.rs | 3 + .../src/http_server/run_http_server.rs | 72 +++++ .../src/job/email_types/mod.rs | 1 + .../password_reset_email_sender.rs | 96 ++++++ .../src/job/job_loop/main_loop.rs | 212 ++++++++++++++ .../email_sender_job/src/job/job_loop/mod.rs | 4 + .../src/job/job_loop/process_single_job.rs | 105 +++++++ .../job/job_loop/process_single_job_error.rs | 18 ++ .../process_single_job_success_case.rs | 14 + .../job/email_sender_job/src/job/mod.rs | 2 + .../email_sender_job/src/job_dependencies.rs | 78 +++++ .../service/job/email_sender_job/src/main.rs | 187 ++++++++++++ .../password_reset_request_handler.rs | 93 +++--- service_cpu.Dockerfile | 9 +- 49 files changed, 2239 insertions(+), 59 deletions(-) create mode 100644 _sql/migrations/2023-11-15-120921_email_sender_jobs/down.sql create mode 100644 _sql/migrations/2023-11-15-120921_email_sender_jobs/up.sql create mode 100644 crates/_types/enums/src/by_table/email_sender_jobs/email_category.rs create mode 100644 crates/_types/enums/src/by_table/email_sender_jobs/mod.rs create mode 100644 crates/_types/tokens/src/tokens/email_sender_job_token.rs create mode 100644 crates/lib/mysql_queries/src/payloads/email_sender_jobs/email_sender_job_args.rs create mode 100644 crates/lib/mysql_queries/src/payloads/email_sender_jobs/mod.rs create mode 100644 crates/lib/mysql_queries/src/payloads/email_sender_jobs/subtypes/email_job_password_reset_args.rs create mode 100644 crates/lib/mysql_queries/src/payloads/email_sender_jobs/subtypes/email_job_welcome_args.rs create mode 100644 crates/lib/mysql_queries/src/payloads/email_sender_jobs/subtypes/mod.rs create mode 100644 crates/lib/mysql_queries/src/queries/email_sender_jobs/_keys.rs create mode 100644 crates/lib/mysql_queries/src/queries/email_sender_jobs/insert_email_sender_job.rs create mode 100644 crates/lib/mysql_queries/src/queries/email_sender_jobs/list_available_email_sender_jobs.rs create mode 100644 crates/lib/mysql_queries/src/queries/email_sender_jobs/mark_email_sender_job_completely_failed.rs create mode 100644 crates/lib/mysql_queries/src/queries/email_sender_jobs/mark_email_sender_job_failure.rs create mode 100644 crates/lib/mysql_queries/src/queries/email_sender_jobs/mark_email_sender_job_pending_and_grab_lock.rs create mode 100644 crates/lib/mysql_queries/src/queries/email_sender_jobs/mark_email_sender_job_successfully_done.rs create mode 100644 crates/lib/mysql_queries/src/queries/email_sender_jobs/mod.rs create mode 100644 crates/service/job/email_sender_job/Cargo.toml create mode 100644 crates/service/job/email_sender_job/config/email-sender-job.common.env create mode 100644 crates/service/job/email_sender_job/config/email-sender-job.development.env create mode 100644 crates/service/job/email_sender_job/config/email-sender-job.production.env create mode 100644 crates/service/job/email_sender_job/src/http_server/endpoints/health_check_handler.rs create mode 100644 crates/service/job/email_sender_job/src/http_server/endpoints/mod.rs create mode 100644 crates/service/job/email_sender_job/src/http_server/http_server_shared_state.rs create mode 100644 crates/service/job/email_sender_job/src/http_server/mod.rs create mode 100644 crates/service/job/email_sender_job/src/http_server/run_http_server.rs create mode 100644 crates/service/job/email_sender_job/src/job/email_types/mod.rs create mode 100644 crates/service/job/email_sender_job/src/job/email_types/password_reset_email_sender.rs create mode 100644 crates/service/job/email_sender_job/src/job/job_loop/main_loop.rs create mode 100644 crates/service/job/email_sender_job/src/job/job_loop/mod.rs create mode 100644 crates/service/job/email_sender_job/src/job/job_loop/process_single_job.rs create mode 100644 crates/service/job/email_sender_job/src/job/job_loop/process_single_job_error.rs create mode 100644 crates/service/job/email_sender_job/src/job/job_loop/process_single_job_success_case.rs create mode 100644 crates/service/job/email_sender_job/src/job/mod.rs create mode 100644 crates/service/job/email_sender_job/src/job_dependencies.rs create mode 100644 crates/service/job/email_sender_job/src/main.rs diff --git a/Cargo.lock b/Cargo.lock index 556010290a..5caf7caf3b 100644 --- a/Cargo.lock +++ b/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]] diff --git a/Cargo.toml b/Cargo.toml index d6b0d5fe79..7bf961b20f 100644 --- a/Cargo.toml +++ b/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", diff --git a/_sql/migrations/2023-11-15-120921_email_sender_jobs/down.sql b/_sql/migrations/2023-11-15-120921_email_sender_jobs/down.sql new file mode 100644 index 0000000000..b7c8c6d4db --- /dev/null +++ b/_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; diff --git a/_sql/migrations/2023-11-15-120921_email_sender_jobs/up.sql b/_sql/migrations/2023-11-15-120921_email_sender_jobs/up.sql new file mode 100644 index 0000000000..1c80bd272b --- /dev/null +++ b/_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; diff --git a/crates/_types/enums/Cargo.toml b/crates/_types/enums/Cargo.toml index 7b7dbb7dcc..e295f6dd26 100644 --- a/crates/_types/enums/Cargo.toml +++ b/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" diff --git a/crates/_types/enums/src/by_table/email_sender_jobs/email_category.rs b/crates/_types/enums/src/by_table/email_sender_jobs/email_category.rs new file mode 100644 index 0000000000..87fedc8144 --- /dev/null +++ b/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 { + match value { + "welcome" => Ok(Self::Welcome), + "password_reset" => Ok(Self::PasswordReset), + _ => Err(format!("invalid value: {:?}", value)), + } + } + + pub fn all_variants() -> BTreeSet { + // NB: BTreeSet is sorted + // NB: BTreeSet::from() isn't const, but not worth using LazyStatic, etc. + BTreeSet::from([ + Self::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()); + } + } + } +} diff --git a/crates/_types/enums/src/by_table/email_sender_jobs/mod.rs b/crates/_types/enums/src/by_table/email_sender_jobs/mod.rs new file mode 100644 index 0000000000..18d5731795 --- /dev/null +++ b/crates/_types/enums/src/by_table/email_sender_jobs/mod.rs @@ -0,0 +1 @@ +pub mod email_category; diff --git a/crates/_types/enums/src/by_table/mod.rs b/crates/_types/enums/src/by_table/mod.rs index 561c3da6f7..1061281e12 100644 --- a/crates/_types/enums/src/by_table/mod.rs +++ b/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; diff --git a/crates/_types/enums/src/common/job_status_plus.rs b/crates/_types/enums/src/common/job_status_plus.rs index 9d55939b75..4f69359e73 100644 --- a/crates/_types/enums/src/common/job_status_plus.rs +++ b/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` diff --git a/crates/_types/tokens/src/prefixes.rs b/crates/_types/tokens/src/prefixes.rs index a0ff368208..41b6217e91 100644 --- a/crates/_types/tokens/src/prefixes.rs +++ b/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; diff --git a/crates/_types/tokens/src/tokens/email_sender_job_token.rs b/crates/_types/tokens/src/tokens/email_sender_job_token.rs new file mode 100644 index 0000000000..359c3c2239 --- /dev/null +++ b/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); diff --git a/crates/_types/tokens/src/tokens/mod.rs b/crates/_types/tokens/src/tokens/mod.rs index 62b55b3e17..1777a47e2f 100644 --- a/crates/_types/tokens/src/tokens/mod.rs +++ b/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; diff --git a/crates/lib/mysql_queries/src/payloads/email_sender_jobs/email_sender_job_args.rs b/crates/lib/mysql_queries/src/payloads/email_sender_jobs/email_sender_job_args.rs new file mode 100644 index 0000000000..906b9bd185 --- /dev/null +++ b/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, +} + +// 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 { + Ok(serde_json::from_str(json)?) + } + + pub fn to_json(&self) -> AnyhowResult { + Ok(serde_json::to_string(self)?) + } +} diff --git a/crates/lib/mysql_queries/src/payloads/email_sender_jobs/mod.rs b/crates/lib/mysql_queries/src/payloads/email_sender_jobs/mod.rs new file mode 100644 index 0000000000..a6e6e22049 --- /dev/null +++ b/crates/lib/mysql_queries/src/payloads/email_sender_jobs/mod.rs @@ -0,0 +1,2 @@ +pub mod email_sender_job_args; +pub mod subtypes; \ No newline at end of file diff --git a/crates/lib/mysql_queries/src/payloads/email_sender_jobs/subtypes/email_job_password_reset_args.rs b/crates/lib/mysql_queries/src/payloads/email_sender_jobs/subtypes/email_job_password_reset_args.rs new file mode 100644 index 0000000000..6b8390accb --- /dev/null +++ b/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, +} + +#[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); + } +} diff --git a/crates/lib/mysql_queries/src/payloads/email_sender_jobs/subtypes/email_job_welcome_args.rs b/crates/lib/mysql_queries/src/payloads/email_sender_jobs/subtypes/email_job_welcome_args.rs new file mode 100644 index 0000000000..b69335a4ca --- /dev/null +++ b/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 { + +} diff --git a/crates/lib/mysql_queries/src/payloads/email_sender_jobs/subtypes/mod.rs b/crates/lib/mysql_queries/src/payloads/email_sender_jobs/subtypes/mod.rs new file mode 100644 index 0000000000..a6716c0385 --- /dev/null +++ b/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; diff --git a/crates/lib/mysql_queries/src/payloads/mod.rs b/crates/lib/mysql_queries/src/payloads/mod.rs index e93c4fd22f..827347da83 100644 --- a/crates/lib/mysql_queries/src/payloads/mod.rs +++ b/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; diff --git a/crates/lib/mysql_queries/src/queries/email_sender_jobs/_keys.rs b/crates/lib/mysql_queries/src/queries/email_sender_jobs/_keys.rs new file mode 100644 index 0000000000..d5f1f2c013 --- /dev/null +++ b/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); diff --git a/crates/lib/mysql_queries/src/queries/email_sender_jobs/insert_email_sender_job.rs b/crates/lib/mysql_queries/src/queries/email_sender_jobs/insert_email_sender_job.rs new file mode 100644 index 0000000000..b54718b211 --- /dev/null +++ b/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, + + 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)) +} diff --git a/crates/lib/mysql_queries/src/queries/email_sender_jobs/list_available_email_sender_jobs.rs b/crates/lib/mysql_queries/src/queries/email_sender_jobs/list_available_email_sender_jobs.rs new file mode 100644 index 0000000000..30436fe877 --- /dev/null +++ b/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, + + // Email details + pub destination_email_address: String, + pub maybe_destination_user_token: Option, + + // Language details + pub ietf_language_tag: Option, + pub ietf_primary_language_subtag: Option, + + // User information to propagate downstream + pub maybe_creator_user_token: Option, + pub maybe_creator_anonymous_visitor_token: Option, + 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, + + pub created_at: chrono::DateTime, + pub updated_at: chrono::DateTime, + pub retry_at: Option>, +} + +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> +{ + 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 = 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> + // https://stackoverflow.com/a/63798748 + .into_iter() + .collect::, anyhow::Error>>()?; + + Ok(job_records) +} + +async fn list_sorted_by_id(args: ListAvailableEmailSenderJobArgs<'_>) -> Result, 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, 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, + + // Email details + pub destination_email_address: String, + pub maybe_destination_user_token: Option, + + // Language details + pub ietf_language_tag: Option, + pub ietf_primary_language_subtag: Option, + + // User information to propagate downstream + pub maybe_creator_user_token: Option, + pub maybe_creator_anonymous_visitor_token: Option, + 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, + + pub created_at: chrono::DateTime, + pub updated_at: chrono::DateTime, + pub retry_at: Option>, +} diff --git a/crates/lib/mysql_queries/src/queries/email_sender_jobs/mark_email_sender_job_completely_failed.rs b/crates/lib/mysql_queries/src/queries/email_sender_jobs/mark_email_sender_job_completely_failed.rs new file mode 100644 index 0000000000..9b6b9b9fd0 --- /dev/null +++ b/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(()), + } +} diff --git a/crates/lib/mysql_queries/src/queries/email_sender_jobs/mark_email_sender_job_failure.rs b/crates/lib/mysql_queries/src/queries/email_sender_jobs/mark_email_sender_job_failure.rs new file mode 100644 index 0000000000..985c5fbc2e --- /dev/null +++ b/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(()), + } +} diff --git a/crates/lib/mysql_queries/src/queries/email_sender_jobs/mark_email_sender_job_pending_and_grab_lock.rs b/crates/lib/mysql_queries/src/queries/email_sender_jobs/mark_email_sender_job_pending_and_grab_lock.rs new file mode 100644 index 0000000000..f4c33c40ae --- /dev/null +++ b/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 { + + // 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) +} diff --git a/crates/lib/mysql_queries/src/queries/email_sender_jobs/mark_email_sender_job_successfully_done.rs b/crates/lib/mysql_queries/src/queries/email_sender_jobs/mark_email_sender_job_successfully_done.rs new file mode 100644 index 0000000000..802e414e82 --- /dev/null +++ b/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(()), + } +} diff --git a/crates/lib/mysql_queries/src/queries/email_sender_jobs/mod.rs b/crates/lib/mysql_queries/src/queries/email_sender_jobs/mod.rs new file mode 100644 index 0000000000..534020aa7f --- /dev/null +++ b/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; diff --git a/crates/lib/mysql_queries/src/queries/mod.rs b/crates/lib/mysql_queries/src/queries/mod.rs index 9fc44e3469..523ec28d2f 100644 --- a/crates/lib/mysql_queries/src/queries/mod.rs +++ b/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; diff --git a/crates/lib/mysql_queries/src/queries/users/user_password_resets/create_password_reset_request.rs b/crates/lib/mysql_queries/src/queries/users/user_password_resets/create_password_reset_request.rs index c534c49803..5ec39b5a63 100644 --- a/crates/lib/mysql_queries/src/queries/users/user_password_resets/create_password_reset_request.rs +++ b/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!( diff --git a/crates/service/job/email_sender_job/Cargo.toml b/crates/service/job/email_sender_job/Cargo.toml new file mode 100644 index 0000000000..eda6ac99c5 --- /dev/null +++ b/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 ", + "Brandon Thomas ", +] +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" diff --git a/crates/service/job/email_sender_job/config/email-sender-job.common.env b/crates/service/job/email_sender_job/config/email-sender-job.common.env new file mode 100644 index 0000000000..f54d58c915 --- /dev/null +++ b/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) diff --git a/crates/service/job/email_sender_job/config/email-sender-job.development.env b/crates/service/job/email_sender_job/config/email-sender-job.development.env new file mode 100644 index 0000000000..8ded4961c5 --- /dev/null +++ b/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" diff --git a/crates/service/job/email_sender_job/config/email-sender-job.production.env b/crates/service/job/email_sender_job/config/email-sender-job.production.env new file mode 100644 index 0000000000..f54d58c915 --- /dev/null +++ b/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) diff --git a/crates/service/job/email_sender_job/src/http_server/endpoints/health_check_handler.rs b/crates/service/job/email_sender_job/src/http_server/endpoints/health_check_handler.rs new file mode 100644 index 0000000000..a77d9b1061 --- /dev/null +++ b/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> +) -> Result { + 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)) + } +} diff --git a/crates/service/job/email_sender_job/src/http_server/endpoints/mod.rs b/crates/service/job/email_sender_job/src/http_server/endpoints/mod.rs new file mode 100644 index 0000000000..0f408270bf --- /dev/null +++ b/crates/service/job/email_sender_job/src/http_server/endpoints/mod.rs @@ -0,0 +1 @@ +pub mod health_check_handler; diff --git a/crates/service/job/email_sender_job/src/http_server/http_server_shared_state.rs b/crates/service/job/email_sender_job/src/http_server/http_server_shared_state.rs new file mode 100644 index 0000000000..534526580f --- /dev/null +++ b/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, +} diff --git a/crates/service/job/email_sender_job/src/http_server/mod.rs b/crates/service/job/email_sender_job/src/http_server/mod.rs new file mode 100644 index 0000000000..5d888824cc --- /dev/null +++ b/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; diff --git a/crates/service/job/email_sender_job/src/http_server/run_http_server.rs b/crates/service/job/email_sender_job/src/http_server/run_http_server.rs new file mode 100644 index 0000000000..6114dc6802 --- /dev/null +++ b/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 +{ + // 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(()) +} diff --git a/crates/service/job/email_sender_job/src/job/email_types/mod.rs b/crates/service/job/email_sender_job/src/job/email_types/mod.rs new file mode 100644 index 0000000000..352a7caee5 --- /dev/null +++ b/crates/service/job/email_sender_job/src/job/email_types/mod.rs @@ -0,0 +1 @@ +pub mod password_reset_email_sender; \ No newline at end of file diff --git a/crates/service/job/email_sender_job/src/job/email_types/password_reset_email_sender.rs b/crates/service/job/email_sender_job/src/job/email_types/password_reset_email_sender.rs new file mode 100644 index 0000000000..e790c7bf35 --- /dev/null +++ b/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 " + .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#" + Click here to reset your password! +
+
+ If you can't click the link, here's the secret reset code: {secret_key} +
+
+ Thank You, +
+
+ 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(()) +} diff --git a/crates/service/job/email_sender_job/src/job/job_loop/main_loop.rs b/crates/service/job/email_sender_job/src/job/job_loop/main_loop.rs new file mode 100644 index 0000000000..a6536c62b7 --- /dev/null +++ b/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) -> 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(()) +} diff --git a/crates/service/job/email_sender_job/src/job/job_loop/mod.rs b/crates/service/job/email_sender_job/src/job/job_loop/mod.rs new file mode 100644 index 0000000000..4a19a57f2a --- /dev/null +++ b/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; \ No newline at end of file diff --git a/crates/service/job/email_sender_job/src/job/job_loop/process_single_job.rs b/crates/service/job/email_sender_job/src/job/job_loop/process_single_job.rs new file mode 100644 index 0000000000..3579ceba81 --- /dev/null +++ b/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 { + 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 { + + 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 { + + 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) +} diff --git a/crates/service/job/email_sender_job/src/job/job_loop/process_single_job_error.rs b/crates/service/job/email_sender_job/src/job/job_loop/process_single_job_error.rs new file mode 100644 index 0000000000..a96550ab26 --- /dev/null +++ b/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 +} diff --git a/crates/service/job/email_sender_job/src/job/job_loop/process_single_job_success_case.rs b/crates/service/job/email_sender_job/src/job/job_loop/process_single_job_success_case.rs new file mode 100644 index 0000000000..94ee3f547b --- /dev/null +++ b/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, +} diff --git a/crates/service/job/email_sender_job/src/job/mod.rs b/crates/service/job/email_sender_job/src/job/mod.rs new file mode 100644 index 0000000000..08cf4058cb --- /dev/null +++ b/crates/service/job/email_sender_job/src/job/mod.rs @@ -0,0 +1,2 @@ +pub mod email_types; +pub mod job_loop; diff --git a/crates/service/job/email_sender_job/src/job_dependencies.rs b/crates/service/job/email_sender_job/src/job_dependencies.rs new file mode 100644 index 0000000000..de6a9d887e --- /dev/null +++ b/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>, + + pub email_sender: SmtpEmailSender, + + pub server_environment: ServerEnvironment, + + pub maybe_keepalive_redis_pool: Option>, + + pub job_progress_reporter: Box, + + 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, +} + +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, +} diff --git a/crates/service/job/email_sender_job/src/main.rs b/crates/service/job/email_sender_job/src/main.rs new file mode 100644 index 0000000000..48784878fb --- /dev/null +++ b/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; + + 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(()) +} diff --git a/crates/service/plugins/users/src/endpoints/password_reset_request_handler.rs b/crates/service/plugins/users/src/endpoints/password_reset_request_handler.rs index f39bec371f..42ea6b900e 100644 --- a/crates/service/plugins/users/src/endpoints/password_reset_request_handler.rs +++ b/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, mysql_pool: web::Data, - server_environment: web::Data, - sender: web::Data, + _server_environment: web::Data, + _sender: web::Data, ) -> Result { 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 " - .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#" - Click here to reset your password! -
-
- If you can't click the link, here's the secret reset code: {secret_key} -
-
- Thank You, -
-
- 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 { let response = PasswordResetRequestedResponse { success: true, diff --git a/service_cpu.Dockerfile b/service_cpu.Dockerfile index 9db0bb7245..92ff685233 100644 --- a/service_cpu.Dockerfile +++ b/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 /