email with token & confirmation works

Cargo.toml

@@ -32,6 +32,7 @@ unicode-segmentation = "1"
 claim = "0.5"
 validator = "0.16"
 lettre = "0.11"
+rand = { version = "0.8", features = ["std_rng"] }
 
 [dev-dependencies]
 fake = "2.9"

configuration/development.yaml

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

src/configuration.rs

@@ -15,6 +15,7 @@ pub struct Settings {
 pub struct ApplicationSettings {
 	pub port: u16,
 	pub host: String,
+	pub base_url: String,
 }
 
 #[derive(serde::Deserialize,Clone)]

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

@@ -1,4 +1,5 @@
 use actix_web::{web, HttpResponse};
+use rand::{distributions::Alphanumeric, thread_rng, Rng};
 use serde::Deserialize;
 use chrono::Utc;
 use uuid::Uuid;
@@ -7,7 +8,7 @@ use lettre::{
     address::AddressError, message::{header::ContentType, Mailbox}, transport::smtp::authentication::Credentials, Message, SmtpTransport, Transport
 };
 
-use crate::{domain::{NewSubscriber, SubscriberEmail, SubscriberName}, email_client::EmailClient};
+use crate::{domain::{NewSubscriber, SubscriberEmail, SubscriberName}, email_client::EmailClient, startup::ApplicationBaseUrl};
 
 #[derive(Deserialize)]
 pub struct FormData {
@@ -17,7 +18,7 @@ pub struct FormData {
 
 #[tracing::instrument(
 	name = "Adding a new subscriber",
-	skip(form, connection_pool, email_client),
+	skip(form, connection_pool, email_client, base_url),
 	fields(
 		subscriber_email = %form.email,
 		subscriber_name = %form.name
@@ -27,16 +28,21 @@ pub async fn subscribe(
 	form: web::Form<FormData>,
 	connection_pool: web::Data<Pool<Postgres>>,
 	email_client: web::Data<EmailClient>,
+	base_url: web::Data<ApplicationBaseUrl>,
 ) -> HttpResponse {
 	let new_subscriber = match form.0.try_into() {
 		Ok(subscriber) => subscriber,
 		Err(_) => return HttpResponse::BadRequest().finish(),
 	};
-	match insert_subscriber(&new_subscriber, &connection_pool).await {
+	let subscriber_id = match insert_subscriber(&new_subscriber, &connection_pool).await {
-		Ok(_) => (),
+		Ok(subscriber_id) => subscriber_id,
 		Err(_) => return HttpResponse::InternalServerError().finish(),
+	};
+	let subscription_token = generate_confirmation_token();
+	if store_token(&connection_pool, &subscriber_id, &subscription_token).await.is_err() {
+		return HttpResponse::InternalServerError().finish();
 	}
-	if send_confirmation_email(&email_client, new_subscriber).await.is_err() {
+	if send_confirmation_email(&email_client, new_subscriber, &base_url.0, &subscription_token).await.is_err() {
 		return HttpResponse::InternalServerError().finish();
 	}
 	HttpResponse::Ok().finish()
@@ -44,13 +50,15 @@ pub async fn subscribe(
 
 #[tracing::instrument(
 	name = "Send a confirmation email to the new subscriber",
-	skip(email_client, new_subscriber)
+	skip(email_client, new_subscriber, base_url)
 )]
 pub async fn send_confirmation_email(
 	email_client: &EmailClient,
 	new_subscriber: NewSubscriber,
+	base_url: &str,
+	subscription_token: &str,
 ) -> Result<(), reqwest::Error> {
-	let confirmation_link = "https://my-api.com/subscriptions/confirm";
+	let confirmation_link = format!("{}/subscriptions/confirm?subscription_token={}", base_url, subscription_token);
 	let plain_body = &format!(
 		"Welcome to our newsletter!\n\
 		Visit {} to confirm your subscription.",
@@ -133,13 +141,14 @@ impl TryFrom<FormData> for NewSubscriber {
 	name = "Saving new subscriber details in the database",
 	skip(new_subscriber, connection_pool)
 )]
-async fn insert_subscriber(new_subscriber: &NewSubscriber, connection_pool: &Pool<Postgres>) -> Result<(), sqlx::Error> {
+async fn insert_subscriber(new_subscriber: &NewSubscriber, connection_pool: &Pool<Postgres>) -> Result<Uuid, sqlx::Error> {
+	let subscriber_id = Uuid::new_v4();
 	query!(
 		r#"
 		INSERT INTO subscriptions (id, email, name, subscribed_at, status)
 		VALUES ($1, $2, $3, $4, 'pending_confirmation')
 		"#,
-		Uuid::new_v4(),
+		subscriber_id,
 		new_subscriber.email.as_ref(),
 		new_subscriber.name.as_ref(),
 		Utc::now()
@@ -150,5 +159,35 @@ async fn insert_subscriber(new_subscriber: &NewSubscriber, connection_pool: &Poo
 		tracing::error!("Failed to execute query: {:?}", e);
 		e
 	})?;
+	Ok(subscriber_id)
+}
+
+#[tracing::instrument(
+	name = "Storing subscription token in the database",
+	skip(connection_pool, subscriber_id, subscription_token)
+)]
+async fn store_token(connection_pool: &Pool<Postgres>, subscriber_id: &Uuid, subscription_token: &str) -> Result<(), sqlx::Error> {
+	query!(
+		r#"
+		INSERT INTO subscription_tokens (subscription_token, subscriber_id)
+		VALUES ($1, $2)
+		"#,
+		subscription_token,
+		subscriber_id
+	)
+	.execute(connection_pool)
+	.await
+	.map_err(|e| {
+		tracing::error!("Failed to execute query: {:?}", e);
+		e
+	})?;
 	Ok(())
 }
+
+fn generate_confirmation_token() -> String {
+	let mut rng = thread_rng();
+	std::iter::repeat_with(|| rng.sample(Alphanumeric))
+		.map(char::from)
+		.take(25)
+		.collect()
+}

src/routes/subscriptions_confirm.rs

@@ -0,0 +1,73 @@
+use actix_web::{web, HttpResponse};
+use sqlx::{pool::Pool, Postgres};
+use uuid::Uuid;
+
+#[derive(serde::Deserialize)]
+pub struct Parameters {
+	pub subscription_token: String,
+}
+
+#[tracing::instrument(
+	name = "Confirm a pending subscriber",
+	skip(parameters),
+)]
+pub async fn confirm(
+	parameters: web::Query<Parameters>,
+	pool: web::Data<Pool<Postgres>>,
+) -> HttpResponse {
+	let id = match get_subscriber_id_from_token(&pool, &parameters.subscription_token).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 a subscriber as confirmed in the database",
+	skip(pool, subscriber_id),
+)]
+pub async fn confirm_subscriber(
+	pool: &Pool<Postgres>,
+	subscriber_id: Uuid,
+) -> Result<(), sqlx::Error> {
+	sqlx::query!(
+		"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(())
+}
+
+#[tracing::instrument(
+	name = "Retrieve subscriber ID by token from the database",
+	skip(pool, subscription_token),
+)]
+pub async fn get_subscriber_id_from_token(
+	pool: &Pool<Postgres>,
+	subscription_token: &str,
+) -> Result<Option<Uuid>, sqlx::Error> {
+	let result = sqlx::query!(
+		"SELECT subscriber_id FROM subscription_tokens WHERE subscription_token = $1",
+		subscription_token
+	)
+	.fetch_optional(pool)
+	.await
+	.map_err(|e| {
+		tracing::error!("Failed to execute query: {:?}", e);
+		e
+	})?;
+	Ok(result.map(|r| r.subscriber_id))
+}

src/startup.rs

@@ -5,22 +5,26 @@ use tracing_actix_web::TracingLogger;
 use std::net::TcpListener;
 
 use crate::email_client::EmailClient;
-use crate::routes::{health_check, subscribe};
+use crate::routes::{confirm, health_check, subscribe};
 
 pub fn run(
 	listener: TcpListener,
 	connection_pool: Pool<Postgres>,
 	email_client: EmailClient,
+	base_url: String,
 ) -> Result<Server, std::io::Error> {
 	let connection_pool = web::Data::new(connection_pool);
 	let email_client = web::Data::new(email_client);
+	let base_url = web::Data::new(ApplicationBaseUrl(base_url));
     let server = HttpServer::new(move || {
         App::new()
             .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(connection_pool.clone())
             .app_data(email_client.clone())
+            .app_data(base_url.clone())
     })
     .listen(listener)?
     .run();
@@ -50,7 +54,7 @@ impl Application {
 	
 		let listener = TcpListener::bind(format!("{}:{}", config.application.host, config.application.port))?;
 		let port = listener.local_addr().unwrap().port();
-		let server = run(listener, connection_pool, email_client)?;
+		let server = run(listener, connection_pool, email_client, config.application.base_url)?;
 		Ok(Self { port, server })
 	}
 
@@ -63,7 +67,7 @@ impl Application {
 	}
 }
 
-
+pub struct ApplicationBaseUrl(pub String);
 
 pub fn get_connection_pool(config: crate::configuration::DatabaseSettings) -> Pool<Postgres> {
 	Pool::connect_lazy_with(config.with_db())

tests/api/helpers.rs

@@ -4,14 +4,21 @@ use uuid::Uuid;
 use wiremock::MockServer;
 use zero2prod::{configuration::{get_configuration, DatabaseSettings}, startup::{get_connection_pool, Application}, telemetry::{get_subscriber, init_subscriber}};
 
+pub struct ConfirmationLinks {
+	pub html: String,
+	pub plain_text: String,
+}
+
 pub struct TestApp {
 	pub address: String,
 	pub connection_pool: Pool<Postgres>,
 	pub email_server: MockServer,
+	pub port: u16,
 }
 
 impl TestApp {
 	pub async fn post_subscriptions(&self, body: String) -> reqwest::Response {
+		println!("Post Address: {}", &self.address);
 		reqwest::Client::new()
 			.post(&format!("{}/subscriptions", &self.address))
 			.header("Content-Type", "application/x-www-form-urlencoded")
@@ -20,6 +27,24 @@ impl TestApp {
 			.await
 			.expect("Failed to execute request.")
 	}
+
+	pub fn get_confirmation_links(&self, email_request: &wiremock::Request) -> ConfirmationLinks {
+		let body: serde_json::Value = serde_json::from_slice(&email_request.body).unwrap();
+
+		let get_link = |s: &str| {
+			let links: Vec<_> = linkify::LinkFinder::new()
+				.links(s)
+				.filter(|l| *l.kind() == linkify::LinkKind::Url)
+				.collect();
+			assert_eq!(links.len(), 1);
+			links[0].as_str().to_owned()
+		};
+
+		let html = get_link(body["HtmlBody"].as_str().unwrap());
+		let plain_text = get_link(body["TextBody"].as_str().unwrap());
+
+		ConfirmationLinks { html, plain_text }
+	}
 }
 
 static TRACING: Lazy<()> = Lazy::new(|| {
@@ -50,14 +75,16 @@ pub async fn spawn_app() -> TestApp {
 	configure_database(&config.database).await;
 
 	let app = Application::build(config.clone()).await.expect("Failed to build app.");
-	println!("App built at port: {}", app.port());
+	let port = app.port();
 	let address = format!("http://127.0.0.1:{}", app.port());
+	println!("App Address: {}", address);
 	let _fut = tokio::spawn(app.run_until_stopped());
 
 	TestApp {
 		address,
 		connection_pool: get_connection_pool(config.database),
 		email_server,
+		port,
 	}
 }
 

tests/api/main.rs

@@ -1,3 +1,4 @@
 mod helpers;
 mod health_check;
 mod subscriptions;
+mod subscriptions_confirm;

tests/api/subscriptions.rs

@@ -114,20 +114,8 @@ async fn subscribe_sends_a_confirmation_email_with_a_link() {
 	assert_eq!(200, response.status().as_u16());
 
 	let email_request = &app.email_server.received_requests().await.unwrap()[0];
-	let body: serde_json::Value = serde_json::from_slice(&email_request.body).unwrap();
+	let confirmation_links = app.get_confirmation_links(email_request);
-
-	let get_link = |s: &str| {
-		let links: Vec<_> = linkify::LinkFinder::new()
-			.links(s)
-			.filter(|l| *l.kind() == linkify::LinkKind::Url)
-			.collect();
-		assert_eq!(links.len(), 1);
-		links[0].as_str().to_owned()
-	};
-
-	let html_link = get_link(body["HtmlBody"].as_str().unwrap());
-	let text_link = get_link(body["TextBody"].as_str().unwrap());
 
 	// The two links should be identical
-	assert_eq!(html_link, text_link);
+	assert_eq!(confirmation_links.html, confirmation_links.plain_text);
 }

tests/api/subscriptions_confirm.rs

@@ -0,0 +1,46 @@
+use reqwest::Url;
+use wiremock::{matchers::{method, path}, Mock, ResponseTemplate};
+
+use crate::helpers::spawn_app;
+
+#[tokio::test]
+async fn confirmations_without_token_are_rejected_with_a_400() {
+	let app = spawn_app().await;
+	let response = reqwest::get(&format!("{}/subscriptions/confirm", app.address)).await.unwrap();
+	assert_eq!(response.status().as_u16(), 400);
+}
+
+#[tokio::test]
+async fn the_link_returned_by_subscribe_returns_a_200_if_called() {
+	let app = spawn_app().await;
+
+	Mock::given(path("/email"))
+		.and(method("POST"))
+		.respond_with(ResponseTemplate::new(200))
+		.mount(&app.email_server)
+		.await;
+
+	app.post_subscriptions("name=Andre%20Heber&email=andre.heber%40gmx.net".into()).await;
+	let email_request = &app.email_server.received_requests().await.unwrap()[0];
+	let confirmation_links = app.get_confirmation_links(email_request);
+
+	let mut confirmation_link = Url::parse(&confirmation_links.html).unwrap();
+
+	assert_eq!(confirmation_link.host_str().unwrap(), "127.0.0.1");
+	confirmation_link.set_port(Some(app.port)).unwrap();
+
+	let _response = reqwest::get(confirmation_link)
+		.await
+		.unwrap()
+		.error_for_status()
+		.unwrap();
+
+	let saved = sqlx::query!("SELECT email, name, status FROM subscriptions")
+		.fetch_one(&app.connection_pool)
+		.await
+		.expect("Failed to fetch saved subscription");
+
+		assert_eq!(saved.email, "andre.heber@gmx.net");
+		assert_eq!(saved.name, "Andre Heber");
+		assert_eq!(saved.status, "confirmed");
+}