chore: refactoring code for user workflow observability

Author: Utilitycoder

migrations/20230718083246_relax_null_checks_on_idempotency.sql

@@ -1 +1,4 @@
 -- Add migration script here
+ALTER TABLE idempotency ALTER COLUMN response_status_code DROP NOT NULL;
+ALTER TABLE idempotency ALTER COLUMN response_body DROP NOT NULL;
+ALTER TABLE idempotency ALTER COLUMN response_headers DROP NOT NULL;

migrations/20230718135651_create_newsletter_issues_table.sql

@@ -0,0 +1,9 @@
+-- Add migration script here
+CREATE TABLE newsletter_issues (
+    newsletter_issue_id uuid NOT NULL,
+    title TEXT NOT NULL,
+    text_content TEXT NOT NULL,
+    html_content TEXT NOT NULL,
+    published_at TEXT NOT NULL,
+    PRIMARY KEY(newsletter_issue_id)
+);

migrations/20230718140050_create_issue_delivery_queue_table.sql

@@ -0,0 +1,6 @@
+-- Add migration script here
+CREATE TABLE issue_delivery_queue (
+    newsletter_issue_id uuid NOT NULL REFERENCES newsletter_issues (newsletter_issue_id),
+    subscriber_email TEXT NOT NULL,
+    PRIMARY KEY(newsletter_issue_id, subscriber_email)
+);

src/configuration.rs

@@ -36,12 +36,7 @@ impl EmailClientSettings {
     pub fn client(self) -> EmailClient {
         let sender = 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, self.authorization_token, timeout)
     }
     pub fn sender(&self) -> Result<SubscriberEmail, String> {
         SubscriberEmail::parse(self.sender_email.clone())

src/idempotency/mod.rs

@@ -3,3 +3,4 @@ mod persistence;
 pub use key::IdempotencyKey;
 pub use persistence::get_saved_response;
 pub use persistence::save_response;
+pub use persistence::{try_processing, NextAction};

src/idempotency/persistence.rs

@@ -1,9 +1,10 @@
 use super::IdempotencyKey;
 use actix_web::body::to_bytes;
-use actix_web::http::{self, StatusCode};
+use actix_web::http::StatusCode;
 use actix_web::HttpResponse;
 use sqlx::postgres::PgHasArrayType;
 use sqlx::PgPool;
+use sqlx::{Postgres, Transaction};
 use uuid::Uuid;
 
 #[derive(Debug, sqlx::Type)]
@@ -27,9 +28,9 @@ pub async fn get_saved_response(
     let saved_response = sqlx::query!(
         r#"
             SELECT
-            response_status_code,
-            response_headers as "response_headers: Vec<HeaderPairRecord>",
-            response_body
+            response_status_code as "response_status_code!",
+            response_headers as "response_headers!: Vec<HeaderPairRecord>",
+            response_body as "response_body!"
             FROM idempotency
             WHERE
             user_id = $1 AND
@@ -40,6 +41,7 @@ pub async fn get_saved_response(
     )
     .fetch_optional(pool)
     .await?;
+
     if let Some(r) = saved_response {
         let status_code = StatusCode::from_u16(r.response_status_code.try_into()?)?;
         let mut response = HttpResponse::build(status_code);
@@ -53,7 +55,7 @@ pub async fn get_saved_response(
 }
 
 pub async fn save_response(
-    pool: &PgPool,
+    mut transaction: Transaction<'static, Postgres>,
     idempotency_key: &IdempotencyKey,
     user_id: Uuid,
     http_response: HttpResponse,
@@ -72,25 +74,63 @@ pub async fn save_response(
     };
     sqlx::query_unchecked!(
         r#"
-        INSERT INTO idempotency (
-            user_id,
-            idempotency_key,
-            response_status_code,
-            response_headers,
-            response_body,
-            created_at
-            )
-        VALUES ($1, $2, $3, $4, $5, now())
+            UPDATE idempotency
+            SET
+            response_status_code = $3,
+            response_headers = $4,
+            response_body = $5
+            WHERE
+            user_id = $1 AND
+            idempotency_key = $2
         "#,
         user_id,
         idempotency_key.as_ref(),
         status_code,
         headers,
         body.as_ref(),
     )
-    .execute(pool)
+    .execute(&mut transaction)
     .await?;
+    transaction.commit().await?;
 
     let http_response = response_head.set_body(body).map_into_boxed_body();
     Ok(http_response)
 }
+
+pub enum NextAction {
+    StartProcessing(Transaction<'static, Postgres>),
+    ReturnSavedResponse(HttpResponse),
+}
+
+pub async fn try_processing(
+    pool: &PgPool,
+    idempotency_key: &IdempotencyKey,
+    user_id: Uuid,
+) -> Result<NextAction, anyhow::Error> {
+    let mut transaction = pool.begin().await?;
+    let n_inserted_rows = sqlx::query!(
+        r#"
+            INSERT INTO idempotency (
+            user_id,
+            idempotency_key,
+            created_at
+            )
+            VALUES ($1, $2, now())
+            ON CONFLICT DO NOTHING
+        "#,
+        user_id,
+        idempotency_key.as_ref()
+    )
+    .execute(&mut transaction)
+    .await?
+    .rows_affected();
+
+    if n_inserted_rows > 0 {
+        Ok(NextAction::StartProcessing(transaction))
+    } else {
+        let saved_response = get_saved_response(pool, idempotency_key, user_id)
+            .await?
+            .ok_or_else(|| anyhow::anyhow!("We expected a saved response, we didn't find it"))?;
+        Ok(NextAction::ReturnSavedResponse(saved_response))
+    }
+}

src/routes/admin/newsletters/get.rs

@@ -3,8 +3,6 @@ use actix_web::HttpResponse;
 use actix_web_flash_messages::IncomingFlashMessages;
 use std::fmt::Write;
 
-use crate::idempotency;
-
 pub async fn publish_newsletter_form(
     flash_messages: IncomingFlashMessages,
 ) -> Result<HttpResponse, actix_web::Error> {

src/routes/admin/newsletters/post.rs

@@ -1,14 +1,13 @@
 use crate::authentication::UserId;
-use crate::domain::SubscriberEmail;
-use crate::email_client::EmailClient;
-use crate::idempotency::{get_saved_response, save_response};
 use crate::idempotency::IdempotencyKey;
+use crate::idempotency::{get_saved_response, save_response, try_processing, NextAction};
 use crate::utils::{e400, e500, see_other};
 use actix_web::web::ReqData;
 use actix_web::{web, HttpResponse};
 use actix_web_flash_messages::FlashMessage;
 use anyhow::Context;
-use sqlx::PgPool;
+use sqlx::{PgPool, Postgres, Transaction};
+use uuid::Uuid;
 
 #[derive(serde::Deserialize)]
 pub struct FormData {
@@ -20,13 +19,12 @@ pub struct FormData {
 
 #[tracing::instrument(
 name = "Publish a newsletter issue", 
-    skip(form, pool, email_client, user_id),
+    skip_all,
     fields(user_id = %*user_id)
 )]
 pub async fn publish_newsletter(
     form: web::Form<FormData>,
     pool: web::Data<PgPool>,
-    email_client: web::Data<EmailClient>,
     user_id: ReqData<UserId>,
 ) -> Result<HttpResponse, actix_web::Error> {
     let user_id = user_id.into_inner();
@@ -37,65 +35,92 @@ pub async fn publish_newsletter(
         idempotency_key,
     } = form.0;
     let idempotency_key: IdempotencyKey = idempotency_key.try_into().map_err(e400)?;
+    let mut transaction = match try_processing(&pool, &idempotency_key, *user_id)
+        .await
+        .map_err(e500)?
+    {
+        NextAction::StartProcessing(t) => t,
+        NextAction::ReturnSavedResponse(saved_response) => {
+            success_message().send();
+            return Ok(saved_response);
+        }
+    };
+
     if let Some(save_response) = get_saved_response(&pool, &idempotency_key, *user_id)
         .await
         .map_err(e500)?
     {
         return Ok(save_response);
     }
-    let subscribers = get_confirmed_subscribers(&pool).await.map_err(e500)?;
-    for subscriber in subscribers {
-        match subscriber {
-            Ok(subscriber) => {
-                email_client
-                    .send_email(&subscriber.email, &title, &html_content, &text_content)
-                    .await
-                    .with_context(|| {
-                        format!("Failed to send newsletter issue to {}", subscriber.email)
-                    })
-                    .map_err(e500)?;
-            }
+    let issue_id = insert_newsletter_issue(&mut transaction, &title, &text_content, &html_content)
+        .await
+        .context("Failed to store newsletter issue details")
+        .map_err(e500)?;
+    enqueue_delivery_tasks(&mut transaction, issue_id)
+        .await
+        .context("Failed to enqueue delivery task")
+        .map_err(e500)?;
 
-            Err(error) => {
-                tracing::warn!(
-                    error.cause_chain = ?error,
-                    error.message = %error,
-                    "Skipping a confirmed subscriber. \
-                    Their stored contact details are invalid",
-                );
-            }
-        }
-    }
-    FlashMessage::info("The newsletter issue has been published!").send();
     let response = see_other("/admin/newsletters");
-    let response = save_response(&pool, &idempotency_key, *user_id, response)
+    let response = save_response(transaction, &idempotency_key, *user_id, response)
         .await
         .map_err(e500)?;
+    success_message().send();
     Ok(response)
 }
 
-struct ConfirmedSubscriber {
-    email: SubscriberEmail,
+fn success_message() -> FlashMessage {
+    FlashMessage::info("The newsletter issue has been accepted - \
+    emails will go out shortly.",)
+}
+
+#[tracing::instrument(skip_all)]
+async fn insert_newsletter_issue(
+    transaction: &mut Transaction<'_, Postgres>,
+    title: &str,
+    text_content: &str,
+    html_content: &str,
+) -> Result<Uuid, sqlx::Error> {
+    let newsletter_issue_id = Uuid::new_v4();
+    sqlx::query!(
+        r#"
+            INSERT INTO newsletter_issues (
+            newsletter_issue_id,
+            title,
+            text_content,
+            html_content,
+            published_at
+            )
+            VALUES ($1, $2, $3, $4, now())
+        "#,
+        newsletter_issue_id,
+        title,
+        text_content,
+        html_content
+    )
+    .execute(transaction)
+    .await?;
+    Ok(newsletter_issue_id)
 }
 
-#[tracing::instrument(name = "Get confirmed subscribers", skip(pool))]
-async fn get_confirmed_subscribers(
-    pool: &PgPool,
-) -> Result<Vec<Result<ConfirmedSubscriber, anyhow::Error>>, anyhow::Error> {
-    let confirmed_subscribers = sqlx::query!(
+#[tracing::instrument(skip_all)]
+async fn enqueue_delivery_tasks(
+    transaction: &mut Transaction<'_, Postgres>,
+    newsletter_issue_id: Uuid,
+) -> Result<(), sqlx::Error> {
+    sqlx::query!(
         r#"
-        SELECT email
-        FROM subscriptions
-        WHERE status = 'confirmed'
+            INSERT INTO issue_delivery_queue (
+                newsletter_issue_id,
+                subscriber_email
+            )
+            SELECT $1, email
+            FROM subscriptions
+            WHERE status = 'confirmed'
         "#,
+        newsletter_issue_id,
     )
-    .fetch_all(pool)
-    .await?
-    .into_iter()
-    .map(|r| match SubscriberEmail::parse(r.email) {
-        Ok(email) => Ok(ConfirmedSubscriber { email }),
-        Err(error) => Err(anyhow::anyhow!(error)),
-    })
-    .collect();
-    Ok(confirmed_subscribers)
+    .execute(transaction)
+    .await?;
+    Ok(())
 }

src/utils.rs

@@ -1,5 +1,4 @@
 use actix_web::http::header::LOCATION;
-use actix_web::http::StatusCode;
 use actix_web::HttpResponse;
 
 // Return a 400 with the user-representation of the validation error as body.

tests/api/helpers.rs

@@ -3,9 +3,9 @@ use argon2::{Argon2, PasswordHasher};
 use once_cell::sync::Lazy;
 use robust_rust::{
     configuration::{get_configuration, DatabaseSettings},
+    email_client::EmailClient,
     startup::{get_connection_pool, Application},
     telemetry::{get_subscriber, init_subscriber},
-    email_client::EmailClient,
 };
 use sqlx::{Connection, Executor, PgConnection, PgPool};
 use uuid::Uuid;