11.10.4.7

Author: timonviola

src/issue_delivery_worker.rs

@@ -0,0 +1,171 @@
+use std::time::{self, Duration};
+use crate::domain::SubscriberEmail;
+use crate::email_client::{self, EmailClient};
+use crate::{configuration::Settings, startup::get_connection_pool};
+
+use sqlx::{Executor, PgPool, Postgres, Transaction};
+use tracing::{field::display, Span};
+use uuid::Uuid;
+
+enum ExecutionOutcome {
+    TaskCompleted,
+    EmptyQueue,
+}
+
+#[tracing::instrument(
+    skip_all,
+    fields(
+        newsletter_issue_id=tracing::field::Empty,
+        subscriber_email=tracing::field::Empty,
+    ),
+    err
+)]
+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)
+    }
+}
+
+async fn delete_task(
+    mut transaction: PgTransaction,
+    issue_id: Uuid,
+    email: &str,
+) -> Result<(), anyhow::Error> {
+    let query = sqlx::query!(
+        r#"
+        DELETE FROM issue_delivery_queue
+        WHERE
+            newsletter_issue_id = $1 AND
+            subscriber_email = $2
+        "#,
+        issue_id,
+        email
+    );
+    transaction.execute(query).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 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
+    );
+    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,4 +1,7 @@
+use std::fmt::{Debug, Display};
+use tokio::task::JoinError;
 use zero2prod::configuration::get_configuration;
+use zero2prod::issue_delivery_worker::run_worker_until_stopped;
 use zero2prod::startup::Application;
 use zero2prod::telemetry::{get_subscriber, init_subscriber};
 
@@ -8,7 +11,38 @@ async fn main() -> anyhow::Result<()> {
     init_subscriber(subscriber);
 
     let configuration = get_configuration().expect("Failed to read configuration.");
-    let application = Application::build(configuration).await?;
-    application.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
+            )
+        }
+    }
+}