Feat: Implemented a functional email client.

Author: Utilitycoder

src/configuration.rs

@@ -34,10 +34,16 @@ pub struct EmailClientSettings {
 
 impl EmailClientSettings {
     pub fn client(self) -> EmailClient {
-        let sender = self.sender().expect("Invalid sender email address");
+        let sender_email = self.sender().expect("Invalid sender email address");
         let timeout = std::time::Duration::from_millis(self.timeout_ms);
-        EmailClient::new(self.base_url, sender, self.authorization_token, timeout)
+        EmailClient::new(
+            self.base_url,
+            sender_email,
+            self.authorization_token,
+            timeout,
+        )
     }
+
     pub fn sender(&self) -> Result<SubscriberEmail, String> {
         SubscriberEmail::parse(self.sender_email.clone())
     }

src/issue_delivery_worker.rs

@@ -0,0 +1,162 @@
+use crate::domain::SubscriberEmail;
+use crate::email_client::EmailClient;
+use crate::{configuration::Settings, startup::get_connection_pool};
+use sqlx::{PgPool, Postgres, Transaction};
+use std::time::Duration;
+use tracing::{field::display, Span};
+use uuid::Uuid;
+
+pub enum ExecutionOutcome {
+    TaskCompleted,
+    EmptyQueue,
+}
+
+#[tracing::instrument(
+    skip_all,
+    fields(
+        newsletter_issue_id = tracing::field::Empty,
+        subscriber_email = tracing::field::Empty,
+    ),
+    err
+)]
+pub async fn try_execute_task(
+    pool: &PgPool,
+    email_client: &EmailClient,
+) -> Result<ExecutionOutcome, anyhow::Error> {
+    let task = dequeue_task(pool).await?;
+    if task.is_none() {
+        return Ok(ExecutionOutcome::EmptyQueue);
+    }
+
+    let (transaction, issue_id, email) = task.unwrap();
+    Span::current()
+        .record("newsletter_issue_id", &display(issue_id))
+        .record("subscriber_email", &display(&email));
+
+    match SubscriberEmail::parse(email.clone()) {
+        Ok(email) => {
+            let issue = get_issue(pool, issue_id).await?;
+            if let Err(e) = email_client
+                .send_email(
+                    &email,
+                    &issue.title,
+                    &issue.html_content,
+                    &issue.text_content,
+                )
+                .await
+            {
+                tracing::error!(
+                error.cause_chain = ?e,
+                error.message = %e,
+                "Failed to deliver issue to a confirmed subscriber. \
+                Skipping.",
+                );
+            }
+        }
+        Err(e) => {
+            tracing::error!(
+            error.cause_chain = ?e,
+            error.message = %e,
+            "Skipping a confirmed subscriber. \
+            Their stored contact details are invalid",
+            );
+        }
+    }
+
+    delete_task(transaction, issue_id, &email).await?;
+    Ok(ExecutionOutcome::TaskCompleted)
+}
+
+type PgTransaction = Transaction<'static, Postgres>;
+
+#[tracing::instrument(skip_all)]
+async fn dequeue_task(
+    pool: &PgPool,
+) -> Result<Option<(PgTransaction, Uuid, String)>, anyhow::Error> {
+    let mut transaction = pool.begin().await?;
+    let r = sqlx::query!(
+        r#"
+            SELECT newsletter_issue_id, subscriber_email
+            FROM issue_delivery_queue
+            FOR UPDATE
+            SKIP LOCKED
+            LIMIT 1
+        "#,
+    )
+    .fetch_optional(&mut transaction)
+    .await?;
+    if let Some(r) = r {
+        Ok(Some((
+            transaction,
+            r.newsletter_issue_id,
+            r.subscriber_email,
+        )))
+    } else {
+        Ok(None)
+    }
+}
+
+#[tracing::instrument(skip_all)]
+async fn delete_task(
+    mut transaction: PgTransaction,
+    issue_id: Uuid,
+    email: &str,
+) -> Result<(), anyhow::Error> {
+    sqlx::query!(
+        r#"
+        DELETE FROM issue_delivery_queue
+        WHERE
+        newsletter_issue_id = $1 AND
+        subscriber_email = $2
+        "#,
+        issue_id,
+        email
+    )
+    .execute(&mut transaction)
+    .await?;
+    transaction.commit().await?;
+    Ok(())
+}
+
+struct NewsletterIssue {
+    title: String,
+    text_content: String,
+    html_content: String,
+}
+
+#[tracing::instrument(skip_all)]
+async fn get_issue(pool: &PgPool, issue_id: Uuid) -> Result<NewsletterIssue, anyhow::Error> {
+    let issue = sqlx::query_as!(
+        NewsletterIssue,
+        r#"
+        SELECT title, text_content, html_content
+        FROM newsletter_issues
+        WHERE
+        newsletter_issue_id = $1
+        "#,
+        issue_id
+    )
+    .fetch_one(pool)
+    .await?;
+    Ok(issue)
+}
+
+async fn worker_loop(pool: PgPool, email_client: EmailClient) -> Result<(), anyhow::Error> {
+    loop {
+        match try_execute_task(&pool, &email_client).await {
+            Ok(ExecutionOutcome::EmptyQueue) => {
+                tokio::time::sleep(Duration::from_secs(10)).await;
+            }
+            Err(_) => {
+                tokio::time::sleep(Duration::from_secs(1)).await;
+            }
+            Ok(ExecutionOutcome::TaskCompleted) => {}
+        }
+    }
+}
+
+pub async fn run_worker_until_stopped(configuration: Settings) -> Result<(), anyhow::Error> {
+    let connection_pool = get_connection_pool(&configuration.database);
+    let email_client = configuration.email_client.client();
+    worker_loop(connection_pool, email_client).await
+}

src/lib.rs

@@ -3,6 +3,7 @@ pub mod configuration;
 pub mod domain;
 pub mod email_client;
 pub mod idempotency;
+pub mod issue_delivery_worker;
 pub mod routes;
 pub mod session_state;
 pub mod startup;

src/main.rs

@@ -1,14 +1,47 @@
 use robust_rust::{
-    configuration::get_configuration, startup::Application, telemetry::get_subscriber,
-    telemetry::init_subscriber,
+    configuration::get_configuration, issue_delivery_worker::run_worker_until_stopped,
+    startup::Application, telemetry::get_subscriber, telemetry::init_subscriber,
 };
+use std::fmt::{Debug, Display};
+use tokio::task::JoinError;
 
 #[tokio::main]
 async fn main() -> anyhow::Result<()> {
     let subscriber = get_subscriber("robust_rust".into(), "info".into(), std::io::stdout);
     init_subscriber(subscriber);
     let configuration = get_configuration().expect("Failed to read configuration.");
-    let server = Application::build(configuration).await?;
-    server.run_until_stopped().await?;
+    let application = Application::build(configuration.clone()).await?;
+    let application_task = tokio::spawn(application.run_until_stopped());
+    let worker_task = tokio::spawn(run_worker_until_stopped(configuration));
+
+    tokio::select! {
+        o = application_task => report_exit("API", o),
+        o = worker_task => report_exit("Background worker", o),
+    };
+
     Ok(())
 }
+
+fn report_exit(task_name: &str, outcome: Result<Result<(), impl Debug + Display>, JoinError>) {
+    match outcome {
+        Ok(Ok(())) => {
+            tracing::info!("{} has exited", task_name)
+        }
+        Ok(Err(e)) => {
+            tracing::error!(
+            error.cause_chain = ?e,
+            error.message = %e,
+            "{} failed",
+            task_name
+            )
+        }
+        Err(e) => {
+            tracing::error!(
+            error.cause_chain = ?e,
+            error.message = %e,
+            "{}' task failed to complete",
+            task_name
+            )
+        }
+    }
+}

src/routes/admin/newsletters/post.rs

@@ -70,8 +70,10 @@ pub async fn publish_newsletter(
 }
 
 fn success_message() -> FlashMessage {
-    FlashMessage::info("The newsletter issue has been accepted - \
-    emails will go out shortly.",)
+    FlashMessage::info(
+        "The newsletter issue has been accepted - \
+    emails will go out shortly.",
+    )
 }
 
 #[tracing::instrument(skip_all)]

src/startup.rs

@@ -28,18 +28,7 @@ impl Application {
     pub async fn build(configuration: Settings) -> Result<Self, anyhow::Error> {
         let connection_pool = get_connection_pool(&configuration.database);
 
-        let sender_email = configuration
-            .email_client
-            .sender()
-            .expect("Invalid sender email address.");
-
-        let timeout = configuration.email_client.timeout();
-        let email_client = EmailClient::new(
-            configuration.email_client.base_url,
-            sender_email,
-            configuration.email_client.authorization_token,
-            timeout,
-        );
+        let email_client = configuration.email_client.client();
 
         let address = format!(
             "{}:{}",

tests/api/helpers.rs

@@ -4,6 +4,7 @@ use once_cell::sync::Lazy;
 use robust_rust::{
     configuration::{get_configuration, DatabaseSettings},
     email_client::EmailClient,
+    issue_delivery_worker::{try_execute_task, ExecutionOutcome},
     startup::{get_connection_pool, Application},
     telemetry::{get_subscriber, init_subscriber},
 };
@@ -41,6 +42,17 @@ pub struct ConfirmationLinks {
 }
 
 impl TestApp {
+    pub async fn dispatch_all_pending_emails(&self) {
+        loop {
+            if let ExecutionOutcome::EmptyQueue =
+                try_execute_task(&self.db_pool, &self.email_client)
+                    .await
+                    .unwrap()
+            {
+                break;
+            }
+        }
+    }
     pub async fn get_login_html(&self) -> String {
         self.api_client
             .get(&format!("{}/login", &self.address))