chore: Tweaked send_newsletters endpoint for idempotency

Author: Utilitycoder

Cargo.lock

@@ -20,7 +20,8 @@ actix-web-lab = "0.15"
 tokio = { version = "1", features = ["macros", "rt-multi-thread"] }
 serde = {version = "1.0.163", features = ["derive"]}
 serde-aux = "4.2.0"
-serde_json = "1" 
+serde_json = "1"
+serde_urlencoded = "0.7.0" 
 config = {version = "0.13", default-features = false, features = ["yaml"] }
 uuid = { version = "1.3.3", features = ["v4", "serde"] }
 chrono = "0.4.15"

Cargo.toml

@@ -0,0 +1,14 @@
+-- Add migration script here
+CREATE TYPE header_pair AS (
+    name TEXT,
+    value BYTEA
+);
+CREATE TABLE idempotency (
+    user_id uuid NOT NULL REFERENCES users(user_id),
+    idempotency_key TEXT NOT NULL,
+    response_status_code SMALLINT NOT NULL,
+    response_headers header_pair[] NOT NULL,
+    response_body BYTEA NOT NULL,
+    created_at timestamptz NOT NULL,
+    PRIMARY KEY(user_id, idempotency_key)
+);

migrations/20230716160608_create_idempotency_table.sql

@@ -0,0 +1 @@
+-- Add migration script here

migrations/20230718083246_relax_null_checks_on_idempotency.sql

@@ -1,4 +1,5 @@
 use crate::domain::SubscriberEmail;
+use crate::email_client::EmailClient;
 use secrecy::{ExposeSecret, Secret};
 use serde_aux::field_attributes::deserialize_number_from_string;
 use sqlx::postgres::{PgConnectOptions, PgSslMode};
@@ -32,6 +33,16 @@ pub struct EmailClientSettings {
 }
 
 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,
+        )
+    }
     pub fn sender(&self) -> Result<SubscriberEmail, String> {
         SubscriberEmail::parse(self.sender_email.clone())
     }

src/configuration.rs

@@ -0,0 +1,31 @@
+#[derive(Debug)]
+pub struct IdempotencyKey(String);
+
+impl TryFrom<String> for IdempotencyKey {
+    type Error = anyhow::Error;
+    fn try_from(s: String) -> Result<Self, Self::Error> {
+        if s.is_empty() {
+            anyhow::bail!("The idempotency key cannot be empty");
+        }
+        let max_length = 50;
+        if s.len() >= max_length {
+            anyhow::bail!(
+                "The idempotency key must be shorter
+than {max_length} characters"
+            );
+        }
+        Ok(Self(s))
+    }
+}
+
+impl From<IdempotencyKey> for String {
+    fn from(k: IdempotencyKey) -> Self {
+        k.0
+    }
+}
+
+impl AsRef<str> for IdempotencyKey {
+    fn as_ref(&self) -> &str {
+        &self.0
+    }
+}

src/idempotency/key.rs

@@ -0,0 +1,5 @@
+mod key;
+mod persistence;
+pub use key::IdempotencyKey;
+pub use persistence::get_saved_response;
+pub use persistence::save_response;

src/idempotency/mod.rs

@@ -0,0 +1,96 @@
+use super::IdempotencyKey;
+use actix_web::body::to_bytes;
+use actix_web::http::{self, StatusCode};
+use actix_web::HttpResponse;
+use sqlx::postgres::PgHasArrayType;
+use sqlx::PgPool;
+use uuid::Uuid;
+
+#[derive(Debug, sqlx::Type)]
+#[sqlx(type_name = "header_pair")]
+struct HeaderPairRecord {
+    name: String,
+    value: Vec<u8>,
+}
+
+impl PgHasArrayType for HeaderPairRecord {
+    fn array_type_info() -> sqlx::postgres::PgTypeInfo {
+        sqlx::postgres::PgTypeInfo::with_name("_header_pair")
+    }
+}
+
+pub async fn get_saved_response(
+    pool: &PgPool,
+    idempotency_key: &IdempotencyKey,
+    user_id: Uuid,
+) -> Result<Option<HttpResponse>, anyhow::Error> {
+    let saved_response = sqlx::query!(
+        r#"
+            SELECT
+            response_status_code,
+            response_headers as "response_headers: Vec<HeaderPairRecord>",
+            response_body
+            FROM idempotency
+            WHERE
+            user_id = $1 AND
+            idempotency_key = $2
+        "#,
+        user_id,
+        idempotency_key.as_ref()
+    )
+    .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);
+        for HeaderPairRecord { name, value } in r.response_headers {
+            response.append_header((name, value));
+        }
+        Ok(Some(response.body(r.response_body)))
+    } else {
+        Ok(None)
+    }
+}
+
+pub async fn save_response(
+    pool: &PgPool,
+    idempotency_key: &IdempotencyKey,
+    user_id: Uuid,
+    http_response: HttpResponse,
+) -> Result<HttpResponse, anyhow::Error> {
+    let (response_head, body) = http_response.into_parts();
+    let body = to_bytes(body).await.map_err(|e| anyhow::anyhow!("{}", e))?;
+    let status_code = response_head.status().as_u16() as i16;
+    let headers = {
+        let mut h = Vec::with_capacity(response_head.headers().len());
+        for (name, value) in response_head.headers().iter() {
+            let name = name.as_str().to_owned();
+            let value = value.as_bytes().to_owned();
+            h.push(HeaderPairRecord { name, value });
+        }
+        h
+    };
+    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())
+        "#,
+        user_id,
+        idempotency_key.as_ref(),
+        status_code,
+        headers,
+        body.as_ref(),
+    )
+    .execute(pool)
+    .await?;
+
+    let http_response = response_head.set_body(body).map_into_boxed_body();
+    Ok(http_response)
+}

src/idempotency/persistence.rs

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

src/lib.rs

@@ -3,13 +3,16 @@ 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> {
     let mut incoming_flash = String::new();
     for message in flash_messages.iter() {
         writeln!(incoming_flash, "<p><i>{}</i></p>", message.content()).unwrap();
     }
+    let idempotency_key = uuid::Uuid::new_v4().to_string();
     Ok(HttpResponse::Ok()
         .content_type(ContentType::html())
         .body(format!(
@@ -48,6 +51,7 @@ pub async fn publish_newsletter_form(
                             ></textarea>
                         </label>
                         <br>
+                        <input hidden type="text" name="idempotency_key" value="{idempotency_key}">
                         <button type="submit">Publish</button>
                     </form>
                     <p><a href="/admin/dashboard">&lt;- Back</a></p>

src/routes/admin/newsletters/get.rs

@@ -1,42 +1,54 @@
 use crate::authentication::UserId;
 use crate::domain::SubscriberEmail;
 use crate::email_client::EmailClient;
-use crate::utils::{e500, see_other};
+use crate::idempotency::{get_saved_response, save_response};
+use crate::idempotency::IdempotencyKey;
+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;
 
 #[derive(serde::Deserialize)]
-pub struct BodyData {
+pub struct FormData {
     title: String,
     text_content: String,
     html_content: String,
+    idempotency_key: String,
 }
 
 #[tracing::instrument(
 name = "Publish a newsletter issue", 
-    skip(body, pool, email_client, user_id),
+    skip(form, pool, email_client, user_id),
     fields(user_id = %*user_id)
 )]
 pub async fn publish_newsletter(
-    body: web::Form<BodyData>,
+    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();
+    let FormData {
+        title,
+        text_content,
+        html_content,
+        idempotency_key,
+    } = form.0;
+    let idempotency_key: IdempotencyKey = idempotency_key.try_into().map_err(e400)?;
+    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,
-                        &body.title,
-                        &body.html_content,
-                        &body.text_content,
-                    )
+                    .send_email(&subscriber.email, &title, &html_content, &text_content)
                     .await
                     .with_context(|| {
                         format!("Failed to send newsletter issue to {}", subscriber.email)
@@ -55,7 +67,11 @@ pub async fn publish_newsletter(
         }
     }
     FlashMessage::info("The newsletter issue has been published!").send();
-    Ok(see_other("/admin/newsletters"))
+    let response = see_other("/admin/newsletters");
+    let response = save_response(&pool, &idempotency_key, *user_id, response)
+        .await
+        .map_err(e500)?;
+    Ok(response)
 }
 
 struct ConfirmedSubscriber {

src/routes/admin/newsletters/post.rs

@@ -1,6 +1,16 @@
 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.
+// The error root cause is preserved for logging purposes.
+pub fn e400<T: std::fmt::Debug + std::fmt::Display>(e: T) -> actix_web::Error
+where
+    T: std::fmt::Debug + std::fmt::Display + 'static,
+{
+    actix_web::error::ErrorBadRequest(e)
+}
+
 // Return an opaque 500 while preserving the error root's cause for logging.
 pub fn e500<T>(e: T) -> actix_web::Error
 where