feat: 第七章代码(三)

Cargo.toml

@@ -41,6 +41,7 @@ validator = "0.18.1"
 [dev-dependencies]
 claims = "0.7.1"
 fake = "3.0.0"
+linkify = "0.10.0"
 mime = "0.3.17"
 once_cell = "1.20.2"
 quickcheck = "1.0.3"

configuration/local.yaml

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

migrations/20241102065508_add_status_to_subscriptions.sql

@@ -0,0 +1,5 @@
+-- Add migration script here
+ALTER TABLE
+    subscriptions
+ADD
+    COLUMN status TEXT NULL;

migrations/20241102065900_make_status_not_null_in_subscriptions.sql

@@ -0,0 +1,18 @@
+-- Add migration script here
+BEGIN;
+
+UPDATE
+    subscriptions
+SET
+    status = 'confirmed'
+WHERE
+    status IS NULL;
+
+ALTER TABLE
+    subscriptions
+ALTER COLUMN
+    status
+SET
+    NOT NULL;
+
+COMMIT;

migrations/20241102070203_create_subscription_tokens_table.sql

@@ -0,0 +1,6 @@
+-- Add migration script here
+CREATE TABLE subscription_tokens(
+    subscription_token TEXT NOT NULL,
+    subscriber_id uuid NOT NULL REFERENCES subscriptions (id),
+    PRIMARY KEY (subscription_token)
+)

src/configuration.rs

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

src/routes/mod.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/subscriptions.rs

@@ -5,12 +5,14 @@ use axum::{
     response::{IntoResponse, Response},
 };
 use chrono::Utc;
+use rand::{distributions::Alphanumeric, thread_rng, Rng};
 use serde::Deserialize;
 use sqlx::PgPool;
 use uuid::Uuid;
 
 use crate::{
     domain::{NewSubscriber, SubscriberEmail, SubscriberName},
+    email_client::EmailClient,
     startup::AppState,
 };
 
@@ -29,23 +31,69 @@ pub async fn subscribe(
         Ok(subscriber) => subscriber,
         Err(_) => return Err(StatusCode::BAD_REQUEST),
     };
-    match insert_subscriber(&state.db_pool, &new_subscriber).await {
-        Ok(_) => Ok((StatusCode::OK).into_response()),
-        Err(_) => Err(StatusCode::INTERNAL_SERVER_ERROR),
+    let subscriber_id = match insert_subscriber(&state.db_pool, &new_subscriber).await {
+        Ok(subscriber_id) => subscriber_id,
+        Err(_) => return Err(StatusCode::INTERNAL_SERVER_ERROR),
+    };
+    let subscription_token = generate_subscription_token();
+    if store_token(&state.db_pool, subscriber_id, &subscription_token)
+        .await
+        .is_err()
+    {
+        return Err(StatusCode::INTERNAL_SERVER_ERROR);
     }
+
+    if send_confirm_email(
+        &state.email_client,
+        new_subscriber,
+        &state.base_url,
+        &subscription_token,
+    )
+    .await
+    .is_err()
+    {
+        return Err(StatusCode::INTERNAL_SERVER_ERROR);
+    };
+    Ok((StatusCode::OK).into_response())
+}
+
+pub async fn send_confirm_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 plain_body = format!(
+        "Welcome to our newsletter!<br /> \
+            Click <a href=\"{}\">here</a> to confirm your subscription",
+        confirmation_link
+    );
+    let html_body = &format!(
+        "Welcome to our newsletter!<br /> \
+            Click <a href=\"{}\">here</a> to confirm your subscription",
+        confirmation_link
+    );
+    email_client
+        .send_email(new_subscriber.email, "Welcome!", html_body, &plain_body)
+        .await
 }
 
 pub async fn insert_subscriber(
     pool: &PgPool,
     new_subscriber: &NewSubscriber,
-) -> Result<(), sqlx::Error> {
+) -> Result<Uuid, sqlx::Error> {
+    let subscriber_id = Uuid::new_v4();
     sqlx::query(
         r#"
-    INSERT INTO subscriptions (id, email, name, subscribed_at)
-    VALUES ($1, $2, $3, $4)
+    INSERT INTO subscriptions (id, email, name, subscribed_at, status)
+    VALUES ($1, $2, $3, $4, 'pending_confirmation')
     "#,
     )
-    .bind(Uuid::new_v4())
+    .bind(subscriber_id)
     .bind(new_subscriber.email.as_ref())
     .bind(new_subscriber.name.as_ref())
     .bind(Utc::now())
@@ -55,7 +103,7 @@ pub async fn insert_subscriber(
         tracing::error!("Failed to execute query: {:?}", e);
         e
     })?;
-    Ok(())
+    Ok(subscriber_id)
 }
 
 impl TryFrom<FormData> for NewSubscriber {
@@ -67,3 +115,33 @@ impl TryFrom<FormData> for NewSubscriber {
         Ok(NewSubscriber { email, name })
     }
 }
+
+fn generate_subscription_token() -> String {
+    let mut rng = thread_rng();
+    std::iter::repeat_with(|| rng.sample(Alphanumeric))
+        .map(char::from)
+        .take(25)
+        .collect()
+}
+
+pub async fn store_token(
+    pool: &PgPool,
+    subscriber_id: Uuid,
+    subscription_token: &str,
+) -> Result<(), sqlx::Error> {
+    sqlx::query(
+        r#"
+    INSERT INTO subscription_tokens (subscription_token,subscriber_id)
+    VALUES ($1, $2)
+    "#,
+    )
+    .bind(subscription_token)
+    .bind(subscriber_id)
+    .execute(pool)
+    .await
+    .map_err(|e| {
+        tracing::error!("Failed to execute query: {:?}", e);
+        e
+    })?;
+    Ok(())
+}

src/routes/subscriptions_confirm.rs

@@ -0,0 +1,78 @@
+use axum::{
+    debug_handler,
+    extract::{Query, State},
+    response::{IntoResponse, Response},
+};
+use reqwest::StatusCode;
+use serde::Deserialize;
+use sqlx::PgPool;
+use uuid::Uuid;
+
+use crate::startup::AppState;
+
+#[derive(Deserialize)]
+pub struct Parameters {
+    subscription_token: String,
+}
+
+#[debug_handler]
+pub async fn confirm(
+    State(state): State<AppState>,
+    Query(params): Query<Parameters>,
+) -> Result<Response, StatusCode> {
+    let id = match get_subscriber_id_from_token(&state.db_pool, &params.subscription_token).await {
+        Ok(id) => id,
+        Err(_) => return Err(StatusCode::INTERNAL_SERVER_ERROR),
+    };
+    dbg!(id);
+    match id {
+        Some(subscriber_id) => {
+            if confirm_subscriber(&state.db_pool, subscriber_id)
+                .await
+                .is_err()
+            {
+                return Err(StatusCode::INTERNAL_SERVER_ERROR);
+            }
+            Ok((StatusCode::OK).into_response())
+        }
+        None => Err(StatusCode::UNAUTHORIZED),
+    }
+}
+
+pub async fn confirm_subscriber(pool: &PgPool, subscriber_id: Uuid) -> Result<(), sqlx::Error> {
+    sqlx::query(
+        r#"
+            UPDATE subscriptions SET status = 'confirmed' WHERE id = $1
+        "#,
+    )
+    .bind(subscriber_id)
+    .execute(pool)
+    .await
+    .map_err(|e| {
+        tracing::error!("Failed to execute query: {:?}", e);
+        e
+    })?;
+    Ok(())
+}
+
+#[derive(sqlx::FromRow, Debug, PartialEq, Eq)]
+pub struct SubscriberId(Uuid);
+
+async fn get_subscriber_id_from_token(
+    pool: &PgPool,
+    subscription_token: &str,
+) -> Result<Option<Uuid>, sqlx::Error> {
+    let result: Option<SubscriberId> = sqlx::query_as(
+        r#"
+        SELECT subscriber_id FROM subscription_tokens WHERE subscription_token=$1
+        "#,
+    )
+    .bind(subscription_token)
+    .fetch_optional(pool)
+    .await
+    .map_err(|e| {
+        tracing::error!("Failed to execute query: {:?}", e);
+        e
+    })?;
+    Ok(result.map(|r| r.0))
+}

src/startup.rs

@@ -11,13 +11,14 @@ use tower_http::trace::TraceLayer;
 use crate::{
     configuration::{DatabaseSettings, Settings},
     email_client::EmailClient,
-    routes::{health, subscribe},
+    routes::{confirm, health, subscribe},
 };
 
 #[derive(Clone)]
 pub struct AppState {
     pub db_pool: Arc<Pool<Postgres>>,
     pub email_client: Arc<EmailClient>,
+    pub base_url: String,
 }
 
 impl AppState {
@@ -42,6 +43,7 @@ impl AppState {
         Self {
             db_pool,
             email_client,
+            base_url: configuration.application.base_url.clone(),
         }
     }
 }
@@ -50,6 +52,7 @@ pub fn app(state: AppState) -> Router {
     Router::new()
         .route("/health", get(health))
         .route("/subscriptions", post(subscribe))
+        .route("/subscriptions/confirm", get(confirm))
         .with_state(state)
         .layer(TraceLayer::new_for_http())
 }

tests/api/health_check.rs

@@ -10,8 +10,8 @@ use crate::helpers::spawn_app;
 
 #[tokio::test]
 async fn health_check_works() {
-    let state = spawn_app().await;
-    let app = app(state);
+    let test_app = spawn_app().await;
+    let app = app(test_app.app_state);
 
     let response = app
         .oneshot(