feat(notifications): add email client (#4002)

* feat(notifications): boilerplate for email client

* chore(deps): add openssl to flake.nix

* chore(notifications): some more boilerplate

* chore(notifications): remove option on EMAIL_PASSWORD

* chore(notifications): remove dependency on openssl

* chore(notifications): add Email as new notification channel

* chore(notifications): rename lettre to smtp

* chore(notifications): remove Email from UserNotificationChannel

* chore(notifications): pass email_executor config in notifications.yml

* chore(notifications): remove password from notifications.yml

* chore(notifications): complete the logic for sending email

* chore(notifications): rename executor -> push_executor in app config

* chore(notifications): provide default value of EMAIL_PASSWORD

* chore(notifications): fix deps

* chore(notifications): do not send emails

* chore(notifications): rename messages

* chore(notifications): add LocalizedEmail

---------

Co-authored-by: Sam Peters <[email protected]>

Cargo.toml

@@ -46,4 +46,5 @@ prost = "0.12"
 rust-i18n = "3"
 google-fcm1 = "5.0.3"
 sqlxmq = { version = "0.5", default-features = false, features = ["runtime-tokio-rustls"] }
+lettre = { version = "0.11.4", default-features = false, features = ["builder", "tokio1", "tokio1-rustls-tls", "smtp-transport"] }
 

core/notifications/BUCK

@@ -69,6 +69,7 @@ galoy_rust_bin(
         "//third-party/rust:prost",
         "//third-party/rust:rust-i18n",
         "//third-party/rust:google-fcm1",
+        "//third-party/rust:lettre",
     ],
     extra_tests = [
         "//lib/tracing-rs:tracing",

core/notifications/Cargo.toml

@@ -44,6 +44,7 @@ tonic-health = { workspace = true }
 prost = { workspace = true }
 rust-i18n = { workspace = true }
 google-fcm1 = { workspace = true }
+lettre = { workspace = true }
 
 [build-dependencies]
 tonic-build = { workspace = true }

core/notifications/notifications.yml

@@ -6,8 +6,13 @@
 # grpc_server:
 #   port: 6685
 app:
-  executor:
+  push_executor:
     fcm:
       google_application_credentials_path: "./config/notifications/fake_service_account.json"
+  email_executor:
+    smtp:
+      username: ""
+      from_email: ""
+      relay: ""
 # kratos_import:
 #   execute_import: true

core/notifications/src/app/config.rs

@@ -1,8 +1,9 @@
 use serde::{Deserialize, Serialize};
 
-use crate::push_executor::PushExecutorConfig;
+use crate::{email_executor::EmailExecutorConfig, push_executor::PushExecutorConfig};
 
 #[derive(Clone, Default, Serialize, Deserialize)]
 pub struct AppConfig {
-    pub executor: PushExecutorConfig,
+    pub push_executor: PushExecutorConfig,
+    pub email_executor: EmailExecutorConfig,
 }

core/notifications/src/app/error.rs

@@ -1,7 +1,8 @@
 use thiserror::Error;
 
 use crate::{
-    job::error::JobError, push_executor::error::PushExecutorError,
+    email_executor::error::EmailExecutorError, job::error::JobError,
+    push_executor::error::PushExecutorError,
     user_notification_settings::error::UserNotificationSettingsError,
 };
 
@@ -14,5 +15,7 @@ pub enum ApplicationError {
     #[error("{0}")]
     PushExecutorError(#[from] PushExecutorError),
     #[error("{0}")]
+    EmailExecutorError(#[from] EmailExecutorError),
+    #[error("{0}")]
     Sqlx(#[from] sqlx::Error),
 }

core/notifications/src/app/mod.rs

@@ -8,7 +8,8 @@ use tracing::instrument;
 use std::sync::Arc;
 
 use crate::{
-    job, notification_event::*, primitives::*, push_executor::*, user_notification_settings::*,
+    email_executor::EmailExecutor, job, notification_event::*, primitives::*, push_executor::*,
+    user_notification_settings::*,
 };
 
 pub use config::*;
@@ -25,8 +26,10 @@ pub struct NotificationsApp {
 impl NotificationsApp {
     pub async fn init(pool: Pool<Postgres>, config: AppConfig) -> Result<Self, ApplicationError> {
         let settings = UserNotificationSettingsRepo::new(&pool);
-        let executor = PushExecutor::init(config.executor.clone(), settings.clone()).await?;
-        let runner = job::start_job_runner(&pool, executor).await?;
+        let push_executor =
+            PushExecutor::init(config.push_executor.clone(), settings.clone()).await?;
+        let email_executor = EmailExecutor::init(config.email_executor.clone(), settings.clone())?;
+        let runner = job::start_job_runner(&pool, push_executor, email_executor).await?;
         Ok(Self {
             _config: config,
             pool,
@@ -173,6 +176,9 @@ impl NotificationsApp {
         event: T,
     ) -> Result<(), ApplicationError> {
         let mut tx = self.pool.begin().await?;
+        if event.should_send_email() {
+            job::spawn_send_email_notification(&mut tx, event.clone().into()).await?;
+        }
         job::spawn_send_push_notification(&mut tx, event.into()).await?;
         tx.commit().await?;
         Ok(())

core/notifications/src/cli/config.rs

@@ -34,13 +34,15 @@ fn default_tracing_config() -> TracingConfig {
 pub struct EnvOverride {
     pub db_con: String,
     pub kratos_pg_con: Option<String>,
+    pub email_password: String,
 }
 
 impl Config {
     pub fn from_path(
         path: Option<impl AsRef<Path>>,
         EnvOverride {
             db_con,
+            email_password,
             kratos_pg_con,
         }: EnvOverride,
     ) -> anyhow::Result<Self> {
@@ -53,8 +55,9 @@ impl Config {
         };
         config.db.pg_con = db_con;
         config.kratos_import.pg_con = kratos_pg_con;
+        config.app.email_executor.smtp.password = email_password;
 
-        config.app.executor.fcm.load_creds()?;
+        config.app.push_executor.fcm.load_creds()?;
 
         Ok(config)
     }