Confirm subscription endpoint

src/routes/subscriptions.rs

@@ -1,63 +1,155 @@
-use crate::domain::{NewSubscriber, SubscriberEmail, SubscriberName};
+use crate::{
+    domain::{NewSubscriber, SubscriberEmail, SubscriberName},
+    email_client::EmailClient,
+    startup::AppState,
+};
 use axum::{Form, extract::State, http::StatusCode, response::IntoResponse};
 use chrono::Utc;
+use rand::{Rng, distr::Alphanumeric};
 use serde::Deserialize;
-use sqlx::PgPool;
+use sqlx::{Executor, Postgres, Transaction};
 use uuid::Uuid;
 
+fn generate_subscription_token() -> String {
+    let mut rng = rand::rng();
+    std::iter::repeat_with(|| rng.sample(Alphanumeric))
+        .map(char::from)
+        .take(25)
+        .collect()
+}
+
 #[tracing::instrument(
     name = "Adding a new subscriber",
-    skip(connection, form),
+    skip(connection_pool, form, email_client),
     fields(
         subscriber_email = %form.email,
         subscriber_name = %form.name
     )
 )]
 pub async fn subscribe(
-    State(connection): State<PgPool>,
+    State(AppState {
+        connection_pool,
+        email_client,
+        base_url,
+    }): State<AppState>,
     Form(form): Form<FormData>,
 ) -> impl IntoResponse {
-    let new_subscriber = match form.try_into() {
-        Ok(subscriber) => subscriber,
-        Err(_) => return StatusCode::BAD_REQUEST,
+    let Ok(mut transaction) = connection_pool.begin().await else {
+        return StatusCode::INTERNAL_SERVER_ERROR;
     };
-    if insert_subscriber(&connection, &new_subscriber)
+    let Ok(new_subscriber) = form.try_into() else {
+        return StatusCode::BAD_REQUEST;
+    };
+    let Ok(subscriber_id) = insert_subscriber(&mut transaction, &new_subscriber).await else {
+        return StatusCode::INTERNAL_SERVER_ERROR;
+    };
+    let subscription_token = generate_subscription_token();
+    if store_token(&mut transaction, &subscription_token, &subscriber_id)
         .await
         .is_err()
     {
-        StatusCode::INTERNAL_SERVER_ERROR
-    } else {
-        StatusCode::OK
+        return StatusCode::INTERNAL_SERVER_ERROR;
     }
+    if send_confirmation_email(
+        &email_client,
+        &new_subscriber,
+        &base_url,
+        &subscription_token,
+    )
+    .await
+    .is_err()
+    {
+        return StatusCode::INTERNAL_SERVER_ERROR;
+    }
+    if transaction.commit().await.is_err() {
+        return StatusCode::INTERNAL_SERVER_ERROR;
+    }
+    StatusCode::OK
 }
 
 #[tracing::instrument(
     name = "Saving new subscriber details in the database",
-    skip(connection, new_subscriber)
+    skip(transaction, new_subscriber)
 )]
 pub async fn insert_subscriber(
-    connection: &PgPool,
+    transaction: &mut Transaction<'_, Postgres>,
     new_subscriber: &NewSubscriber,
-) -> Result<(), sqlx::Error> {
-    sqlx::query!(
+) -> Result<Uuid, sqlx::Error> {
+    let subscriber_id = Uuid::new_v4();
+    let query = 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')
         "#,
-        Uuid::new_v4(),
+        subscriber_id,
         new_subscriber.email.as_ref(),
         new_subscriber.name.as_ref(),
         Utc::now()
-    )
-    .execute(connection)
-    .await
-    .map_err(|e| {
+    );
+    transaction.execute(query).await.map_err(|e| {
+        tracing::error!("Failed to execute query: {:?}", e);
+        e
+    })?;
+    Ok(subscriber_id)
+}
+
+#[tracing::instrument(
+    name = "Store subscription token in the database",
+    skip(transaction, subscription_token)
+)]
+async fn store_token(
+    transaction: &mut Transaction<'_, Postgres>,
+    subscription_token: &str,
+    subscriber_id: &Uuid,
+) -> Result<(), sqlx::Error> {
+    let query = sqlx::query!(
+        r#"
+        INSERT INTO subscription_tokens (subscription_token, subscriber_id)
+        VALUES ($1, $2)
+        "#,
+        subscription_token,
+        subscriber_id,
+    );
+    transaction.execute(query).await.map_err(|e| {
         tracing::error!("Failed to execute query: {:?}", e);
         e
     })?;
     Ok(())
 }
 
+#[tracing::instrument(
+    name = "Send a confirmation email to a new subscriber",
+    skip(email_client, new_subscriber, base_url, subscription_token)
+)]
+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_content = format!(
+        "Welcome to our newsletter!<br />\
+Click <a href=\"{}\">here</a> to confirm your subscription.",
+        confirmation_link
+    );
+    let text_content = format!(
+        "Welcome to our newsletter!\nVisit {} to confirm your subscription.",
+        confirmation_link
+    );
+    email_client
+        .send_email(
+            &new_subscriber.email,
+            "Welcome!",
+            &html_content,
+            &text_content,
+        )
+        .await
+}
+
 #[derive(Debug, Deserialize)]
 #[allow(dead_code)]
 pub struct FormData {