Final push.

src/routes/subscriptions.rs

@@ -43,11 +43,14 @@ pub async fn subscribe(
         .0
         .try_into()
         .map_err(|_| HttpResponse::BadRequest().finish())?;
-    insert_subscriber(&pool, &new_subscriber)
+    let subscriber_id = insert_subscriber(&pool, &new_subscriber)
         .await
         .map_err(|_| HttpResponse::InternalServerError().finish())?;
     // We are swallowing the error for the time being.
     let subscription_token = generate_subscription_token();
+    store_token(&pool, subscriber_id, &subscription_token)
+        .await
+        .map_err(|_| HttpResponse::InternalServerError().finish())?;
     let _ = send_confirmation_email(
         &email_client,
         new_subscriber,
@@ -100,13 +103,14 @@ pub async fn send_confirmation_email(
 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, 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()
@@ -117,5 +121,31 @@ pub async fn insert_subscriber(
         tracing::error!("Failed to execute query: {:?}", e);
         e
     })?;
+    Ok(subscriber_id)
+}
+
+#[tracing::instrument(
+    name = "Store subscription token in the database",
+    skip(subscription_token, pool)
+)]
+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)
+        "#,
+        subscription_token,
+        subscriber_id
+    )
+    .execute(pool)
+    .await
+    .map_err(|e| {
+        tracing::error!("Failed to execute query: {:?}", e);
+        e
+    })?;
     Ok(())
 }

src/routes/subscriptions_confirm.rs

@@ -1,14 +1,61 @@
-use actix_web::{HttpResponse, web};
+use actix_web::{web, HttpResponse};
+use sqlx::PgPool;
+use uuid::Uuid;
 
 #[derive(serde::Deserialize)]
 pub struct Parameters {
-    subscription_token: String
+    subscription_token: String,
 }
 
-#[tracing::instrument(
-    name = "Confirm a pending subscriber",
-    skip(_parameters)
-)]
-pub async fn confirm(_parameters: web::Query<Parameters>) -> HttpResponse {
-    HttpResponse::Ok().finish()
+#[tracing::instrument(name = "Confirm a pending subscriber", skip(parameters, pool))]
+pub async fn confirm(
+    parameters: web::Query<Parameters>,
+    pool: web::Data<PgPool>,
+) -> Result<HttpResponse, HttpResponse> {
+    let id = get_subscriber_id_from_token(&pool, &parameters.subscription_token)
+        .await
+        .map_err(|_| HttpResponse::InternalServerError().finish())?;
+    match id {
+        // Non-existing token!
+        None => Err(HttpResponse::Unauthorized().finish()),
+        Some(subscriber_id) => {
+            confirm_subscriber(&pool, subscriber_id)
+                .await
+                .map_err(|_| HttpResponse::InternalServerError().finish())?;
+            Ok(HttpResponse::Ok().finish())
+        }
+    }
+}
+
+#[tracing::instrument(name = "Mark subscriber as confirmed", skip(subscriber_id, pool))]
+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(())
+}
+
+#[tracing::instrument(name = "Get subscriber_id from token", skip(subscription_token, pool))]
+pub async fn get_subscriber_id_from_token(
+    pool: &PgPool,
+    subscription_token: &str,
+) -> Result<Option<Uuid>, sqlx::Error> {
+    let result = sqlx::query!(
+        r#"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))
 }