chore: Refactor database queries for improved performance

Author: Utilitycoder

Cargo.lock

@@ -30,6 +30,7 @@ sqlx = { version = "0.5.7", default-features = false, features = ["offline", "ru
 unicode-segmentation = "1"
 validator = "0.16.0"
 reqwest = { version = "0.11.18", default-features = false, features = ["json", "rustls-tls"] }
+rand = { version = "0.8", features = ["std_rng"] }
 
 [dev-dependencies]
 once_cell = "1"
@@ -40,6 +41,7 @@ rand = "0.8.5"
 fake = "2.6.1"
 wiremock = "0.5.19"
 serde_json = "1"
+linkify = "0.8.0"
 
 [profile.release]
 strip = true

Cargo.toml

@@ -1,4 +1,5 @@
 application:
   host: 127.0.0.1
+  base_url: "http://127.0.0.1"
 database:
   require_ssl: false

configuration/local.yml

@@ -3,10 +3,6 @@
 -- in more details towards the end of this chapter!
 -- `sqlx` does not do it automatically for us.
 BEGIN;
--- Backfill `status` for historical entries
-UPDATE subscriptions
-SET status = 'confirmed'
-WHERE status IS NULL;
--- Make `status` mandatory
+UPDATE subscriptions SET status = 'confirmed' WHERE status IS NULL;
 ALTER TABLE subscriptions ALTER COLUMN status SET NOT NULL;
 COMMIT;

migrations/20230619112026_make_status_not_null_in_subscriptions.sql

@@ -53,6 +53,9 @@ services:
       - key: APP_DATABASE__DATABASE_NAME
         scope: RUN_TIME
         value: ${newsletter.DATABASE}
+      - key: APP_APPLICATION__BASE_URL
+        scope: RUN_TIME
+        value: ${APP_URL}
 databases:
   # PG = Postgres
   - engine: PG

spec.yaml

@@ -17,6 +17,7 @@ pub struct ApplicationSettings {
     #[serde(deserialize_with = "deserialize_number_from_string")]
     pub port: u16,
     pub host: String,
+    pub base_url: String,
 }
 
 #[derive(serde::Deserialize, Clone)]

src/configuration.rs

@@ -1,5 +1,7 @@
 mod health_check;
 mod subscriptions;
+mod subscriptions_confirm;
 
 pub use health_check::*;
 pub use subscriptions::*;
+pub use subscriptions_confirm::*;

src/routes/mod.rs

@@ -1,9 +1,14 @@
-use crate::domain::{NewSubscriber, SubscriberEmail, SubscriberName};
-use crate::email_client::EmailClient;
+use crate::{
+    domain::{NewSubscriber, SubscriberEmail, SubscriberName},
+    email_client::EmailClient,
+    startup::ApplicationBaseUrl,
+};
 use actix_web::{web, HttpResponse};
 use chrono::Utc;
+use rand::{distributions::Alphanumeric, thread_rng, Rng};
+
 use sqlx::types::Uuid;
-use sqlx::PgPool;
+use sqlx::{PgPool, Postgres, Transaction};
 
 #[allow(dead_code)]
 #[derive(serde::Deserialize)]
@@ -12,9 +17,19 @@ pub struct FormData {
     name: String,
 }
 
+impl TryFrom<FormData> for NewSubscriber {
+    type Error = String;
+
+    fn try_from(value: FormData) -> Result<Self, Self::Error> {
+        let name = SubscriberName::parse(value.name)?;
+        let email = SubscriberEmail::parse(value.email)?;
+        Ok(Self { name, email })
+    }
+}
+
 #[tracing::instrument(
     name = "Adding a new subscriber",
-    skip(form, pool, email_client),
+    skip(form, pool, email_client, base_url),
     fields(
         email = %form.email,
         name = %form.name
@@ -24,61 +39,124 @@ pub async fn subscribe(
     form: web::Form<FormData>,
     pool: web::Data<PgPool>,
     email_client: web::Data<EmailClient>,
+    base_url: web::Data<ApplicationBaseUrl>,
 ) -> HttpResponse {
     let new_subscriber = match form.0.try_into() {
         Ok(subscriber) => subscriber,
         Err(e) => return HttpResponse::BadRequest().body(e),
     };
 
-    if insert_subscriber(&new_subscriber, &pool).await.is_err() {
+    let mut transaction = match pool.begin().await {
+        Ok(t) => t,
+        Err(_) => return HttpResponse::InternalServerError().finish(),
+    };
+
+    let subscriber_id = match insert_subscriber(&new_subscriber, &mut transaction).await {
+        Ok(id) => id,
+        Err(_) => return HttpResponse::InternalServerError().finish(),
+    };
+
+    let subscription_token = generate_subscription_token();
+
+    if store_token(subscriber_id, &subscription_token, &pool).await.is_err() {
         return HttpResponse::InternalServerError().finish();
     }
 
-    if email_client
-        .send_email(
-            new_subscriber.email,
-            "Welcome!",
-            "Welcome to our newsletter!",
-            "Welcome to our newsletter!",
-        )
+    if send_confirmation_email(&email_client, new_subscriber, &base_url.0, &subscription_token)
         .await
         .is_err()
     {
         return HttpResponse::InternalServerError().finish();
     }
 
+    if transaction.commit().await.is_err() {
+        return HttpResponse::InternalServerError().finish();
+    }
+
     HttpResponse::Ok().finish()
 }
 
-impl TryFrom<FormData> for NewSubscriber {
-    type Error = String;
+#[tracing::instrument(
+    name = "Send a confirmation email to a new subscriber",
+    skip(new_subscriber, email_client)
+)]
+pub async fn send_confirmation_email(
+    email_client: &EmailClient,
+    new_subscriber: NewSubscriber,
+    base_url: &str,
+    subscription_token: &str,
+) -> Result<(), reqwest::Error> {
+    let confirmation_link = format!(
+        "{}/subscriptions/confirm?subscription_token={}",
+        base_url, subscription_token
+    );
+    let html_body_text = format!(
+        "Welcome to our newsletter!<br />\
+    Click <a href=\"{}\">here</a> to confirm your subscription.",
+        confirmation_link
+    );
 
-    fn try_from(value: FormData) -> Result<Self, Self::Error> {
-        let name = SubscriberName::parse(value.name)?;
-        let email = SubscriberEmail::parse(value.email)?;
-        Ok(Self { name, email })
-    }
+    let plain_body_text = format!(
+        "Welcome to our newsletter!\nVisit {} to confirm your subscription.",
+        confirmation_link
+    );
+
+    email_client
+        .send_email(
+            new_subscriber.email,
+            "Welcome!",
+            &html_body_text,
+            &plain_body_text,
+        )
+        .await
 }
 
 #[tracing::instrument(
     name = "Saving a new subscriber details in the database",
-    skip(new_subscriber, pool)
+    skip(new_subscriber, transaction),
 )]
 async fn insert_subscriber(
     new_subscriber: &NewSubscriber,
-    pool: &PgPool,
-) -> Result<(), sqlx::Error> {
-    let subscribe_id = Uuid::new_v4();
+    transaction: &mut Transaction<'_, Postgres>
+) -> Result<Uuid, sqlx::Error> {
+    let subscriber_id = Uuid::new_v4();
     sqlx::query!(
         r#"
         INSERT INTO subscriptions (id, email, name, subscribed_at, status)
-        VALUES ($1, $2, $3, $4, 'confirmed')
+        VALUES ($1, $2, $3, $4, 'pending_confirmation')
         "#,
-        subscribe_id,
+        subscriber_id,
         new_subscriber.email.as_ref(),
         new_subscriber.name.as_ref(),
         Utc::now()
     )
+    .execute(transaction)
+    .await
+    .map_err(|e| {
+        tracing::error!("Failed to execute query: {:?}", e);
+        e
+    })?;
+    Ok(subscriber_id)
+}
+
+/// Store subscription token in the database
+#[tracing::instrument(
+    name = "Saving subscription token in the database",
+    skip(subscriber_id, subscription_token, pool)
+)]
+async fn store_token(
+    subscriber_id: Uuid,
+    subscription_token: &str,
+    pool: &PgPool,
+) -> Result<(), sqlx::Error> {
+    sqlx::query!(
+        r#"
+        INSERT INTO subscription_tokens (subscription_token, subscriber_id)
+        VALUES ($1, $2)
+        "#,
+        subscription_token,
+        subscriber_id,
+    )
     .execute(pool)
     .await
     .map_err(|e| {
@@ -87,3 +165,12 @@ async fn insert_subscriber(
     })?;
     Ok(())
 }
+
+/// Gnerate a random 25 character-long case-sensitive subscription token
+fn generate_subscription_token() -> String {
+    let mut rng = thread_rng();
+    std::iter::repeat_with(|| rng.sample(Alphanumeric))
+        .map(char::from)
+        .take(25)
+        .collect()
+}

src/routes/subscriptions.rs

@@ -0,0 +1,56 @@
+use actix_web::{web, HttpResponse};
+use sqlx::{PgPool, types::Uuid};
+
+#[derive(serde::Deserialize)]
+pub struct Parameters {
+    subscription_token: String,
+}
+
+#[tracing::instrument(name = "Confirm a pending subscriber", skip(parameters, pool))]
+pub async fn confirm(parameters: web::Query<Parameters>, pool: web::Data<PgPool>) -> HttpResponse {
+    let id = match get_subscriber_id_from_token(&parameters.subscription_token, &pool).await {
+        Ok(id) => id,
+        Err(_) => return HttpResponse::BadRequest().finish(),
+    };
+    
+    match id {
+        None => HttpResponse::Unauthorized().finish(),
+        Some(subscriber_id) => {
+            if confirm_subscriber(&pool, subscriber_id).await.is_err() {
+                return HttpResponse::InternalServerError().finish();
+            }
+            HttpResponse::Ok().finish()
+        }
+    }
+}
+
+#[tracing::instrument(name = "Mark subscriber as confirmed", skip(pool, subscriber_id))]
+pub async fn confirm_subscriber(pool: &PgPool, subscriber_id: Uuid) -> Result<(), sqlx::Error> {
+    sqlx::query!(
+        r#"UPDATE subscriptions SET status = 'confirmed' WHERE id = $1"#,
+        subscriber_id
+    )
+    .execute(pool)
+    .await
+    .map_err(|e| {
+        tracing::error!("Failed to execute query: {:?}", e);
+        e
+    })?;
+    Ok(())
+}
+
+/// Get subscriber token from the database
+#[tracing::instrument(name = "Getting subscriber ID from token", skip(token, pool))]
+async fn get_subscriber_id_from_token(token: &str, pool: &PgPool) -> Result<Option<Uuid>, sqlx::Error> {
+    let result = sqlx::query!(
+        r#"SELECT subscriber_id FROM subscription_tokens WHERE subscription_token = $1"#,
+        token
+    )
+    .fetch_optional(pool)
+    .await
+    .map_err(|e| {
+        tracing::error!("Failed to execute query: {:?}", e);
+        e
+    })?;
+    Ok(result.map(|r| r.subscriber_id))
+}

src/routes/subscriptions_confirm.rs

@@ -1,7 +1,7 @@
 use crate::{
     configuration::{DatabaseSettings, Settings},
     email_client::EmailClient,
-    routes::{health_check, subscribe},
+    routes::{confirm, health_check, subscribe},
 };
 use actix_web::{dev::Server, web, App, HttpServer};
 use sqlx::postgres::PgPoolOptions;
@@ -38,7 +38,12 @@ impl Application {
 
         let listener = TcpListener::bind(address)?;
         let port = listener.local_addr().unwrap().port();
-        let server = run(listener, connection_pool, email_client)?;
+        let server = run(
+            listener,
+            connection_pool,
+            email_client,
+            configuration.application.base_url,
+        )?;
         Ok(Self { port, server })
     }
 
@@ -57,10 +62,13 @@ pub fn get_connection_pool(configuration: &DatabaseSettings) -> PgPool {
         .connect_lazy_with(configuration.with_db())
 }
 
+pub struct ApplicationBaseUrl(pub String);
+
 pub fn run(
     listener: TcpListener,
     db_pool: PgPool,
     email_client: EmailClient,
+    base_url: String,
 ) -> Result<Server, std::io::Error> {
     let db_pool = web::Data::new(db_pool);
     let email_client = web::Data::new(email_client);
@@ -70,8 +78,10 @@ pub fn run(
             .wrap(TracingLogger::default())
             .route("/health_check", web::get().to(health_check))
             .route("/subscriptions", web::post().to(subscribe))
+            .route("/subscriptions/confirm", web::get().to(confirm))
             .app_data(db_pool.clone())
             .app_data(email_client.clone())
+            .app_data(base_url.clone())
     })
     .listen(listener)?
     .run();