Compute dashboard stats

Track open rate for new post notifications (user clicked the button in
the link or not). No data about the user is collected during the
process, it only uses an ID inserted by the issue delivery worker.

src/issue_delivery_worker.rs

@@ -1,4 +1,7 @@
-use crate::{configuration::Settings, domain::SubscriberEmail, email_client::EmailClient};
+use crate::{
+    configuration::Settings, domain::SubscriberEmail, email_client::EmailClient, routes::EmailType,
+};
+use anyhow::Context;
 use sqlx::{Executor, PgPool, Postgres, Row, Transaction, postgres::PgPoolOptions};
 use std::time::Duration;
 use tracing::{Span, field::display};
@@ -41,10 +44,10 @@ pub async fn try_execute_task(
     email_client: &EmailClient,
 ) -> Result<ExecutionOutcome, anyhow::Error> {
     let task = dequeue_task(connection_pool).await?;
-    if task.is_none() {
-        return Ok(ExecutionOutcome::EmptyQueue);
-    }
-    let (transaction, task) = task.unwrap();
+    let (mut transaction, task) = match task {
+        Some((transaction, task)) => (transaction, task),
+        None => return Ok(ExecutionOutcome::EmptyQueue),
+    };
     Span::current()
         .record("newsletter_issue_id", display(task.newsletter_issue_id))
         .record("subscriber_email", display(&task.subscriber_email));
@@ -52,6 +55,9 @@ pub async fn try_execute_task(
         Ok(email) => {
             let mut issue = get_issue(connection_pool, task.newsletter_issue_id).await?;
             issue.inject_unsubscribe_token(&task.unsubscribe_token);
+            if task.kind == EmailType::NewPost.to_string() {
+                issue.create_tracking_info(&mut transaction).await?;
+            }
             if let Err(e) = email_client
                 .send_email(
                     &email,
@@ -85,6 +91,7 @@ pub async fn try_execute_task(
 }
 
 struct NewsletterIssue {
+    newsletter_issue_id: Uuid,
     title: String,
     text_content: String,
     html_content: String,
@@ -95,6 +102,28 @@ impl NewsletterIssue {
         self.text_content = self.text_content.replace("UNSUBSCRIBE_TOKEN", token);
         self.html_content = self.html_content.replace("UNSUBSCRIBE_TOKEN", token);
     }
+
+    async fn create_tracking_info(
+        &mut self,
+        transaction: &mut Transaction<'static, Postgres>,
+    ) -> Result<(), anyhow::Error> {
+        let email_id = Uuid::new_v4();
+        let query = sqlx::query!(
+            r#"
+            INSERT INTO notifications_delivered (email_id, newsletter_issue_id)
+            VALUES ($1, $2)
+            "#,
+            email_id,
+            self.newsletter_issue_id
+        );
+        transaction
+            .execute(query)
+            .await
+            .context("Failed to store email tracking info.")?;
+        self.text_content = self.text_content.replace("EMAIL_ID", &email_id.to_string());
+        self.html_content = self.html_content.replace("EMAIL_ID", &email_id.to_string());
+        Ok(())
+    }
 }
 
 #[tracing::instrument(skip_all)]
@@ -105,7 +134,7 @@ async fn get_issue(
     let issue = sqlx::query_as!(
         NewsletterIssue,
         r#"
-        SELECT title, text_content, html_content
+        SELECT newsletter_issue_id, title, text_content, html_content
         FROM newsletter_issues
         WHERE newsletter_issue_id = $1
         "#,
@@ -120,6 +149,7 @@ pub struct Task {
     pub newsletter_issue_id: Uuid,
     pub subscriber_email: String,
     pub unsubscribe_token: String,
+    pub kind: String,
 }
 
 #[tracing::instrument(skip_all)]
@@ -129,7 +159,7 @@ async fn dequeue_task(
     let mut transaction = connection_pool.begin().await?;
     let query = sqlx::query!(
         r#"
-        SELECT newsletter_issue_id, subscriber_email, unsubscribe_token
+        SELECT newsletter_issue_id, subscriber_email, unsubscribe_token, kind
         FROM issue_delivery_queue
         FOR UPDATE
         SKIP LOCKED
@@ -142,6 +172,7 @@ async fn dequeue_task(
             newsletter_issue_id: row.get("newsletter_issue_id"),
             subscriber_email: row.get("subscriber_email"),
             unsubscribe_token: row.get("unsubscribe_token"),
+            kind: row.get("kind"),
         };
         Ok(Some((transaction, task)))
     } else {

src/routes/admin/dashboard.rs

@@ -1,20 +1,84 @@
-use crate::{authentication::AuthenticatedUser, templates::DashboardTemplate};
+use crate::{
+    authentication::AuthenticatedUser, routes::AppError, startup::AppState,
+    templates::DashboardTemplate,
+};
+use anyhow::Context;
 use askama::Template;
 use axum::{
     Extension,
+    extract::State,
     response::{Html, IntoResponse, Response},
 };
 use uuid::Uuid;
 
+pub struct DashboardStats {
+    pub subscribers: i64,
+    pub posts: i64,
+    pub notifications_sent: i64,
+    pub open_rate: f64,
+}
+
+impl DashboardStats {
+    pub fn formatted_rate(&self) -> String {
+        format!("{:.1}%", self.open_rate)
+    }
+}
+
 pub async fn admin_dashboard(
+    State(AppState {
+        connection_pool, ..
+    }): State<AppState>,
     Extension(AuthenticatedUser { username, .. }): Extension<AuthenticatedUser>,
-) -> Response {
+) -> Result<Response, AppError> {
+    let stats = get_stats(&connection_pool).await?;
     let idempotency_key_1 = Uuid::new_v4().to_string();
     let idempotency_key_2 = Uuid::new_v4().to_string();
     let template = DashboardTemplate {
         username,
         idempotency_key_1,
         idempotency_key_2,
+        stats,
     };
-    Html(template.render().unwrap()).into_response()
+    Ok(Html(template.render().unwrap()).into_response())
+}
+
+async fn get_stats(connection_pool: &sqlx::PgPool) -> Result<DashboardStats, anyhow::Error> {
+    let subscribers =
+        sqlx::query_scalar!("SELECT count(*) FROM subscriptions WHERE status = 'confirmed'")
+            .fetch_one(connection_pool)
+            .await
+            .context("Failed to fetch subscribers count.")?
+            .unwrap_or(0);
+
+    let posts = sqlx::query_scalar!("SELECT count(*) FROM posts")
+        .fetch_one(connection_pool)
+        .await
+        .context("Failed to fetch posts count.")?
+        .unwrap_or(0);
+
+    let notifications_sent = sqlx::query_scalar!("SELECT count(*) FROM notifications_delivered")
+        .fetch_one(connection_pool)
+        .await
+        .context("Failed to fetch notifications sent count.")?
+        .unwrap_or(0);
+
+    let opened =
+        sqlx::query_scalar!("SELECT count(*) FROM notifications_delivered WHERE opened = TRUE")
+            .fetch_one(connection_pool)
+            .await
+            .context("Failed to fetch notifications sent count.")?
+            .unwrap_or(0);
+
+    let open_rate = if notifications_sent == 0 {
+        0.0
+    } else {
+        (opened as f64) / (notifications_sent as f64) * 100.0
+    };
+
+    Ok(DashboardStats {
+        subscribers,
+        posts,
+        notifications_sent,
+        open_rate,
+    })
 }

src/routes/admin/newsletters.rs

@@ -1,3 +1,5 @@
+use std::fmt::Display;
+
 use crate::{
     authentication::AuthenticatedUser,
     idempotency::{IdempotencyKey, save_response, try_processing},
@@ -46,23 +48,40 @@ pub async fn insert_newsletter_issue(
     Ok(newsletter_issue_id)
 }
 
+pub enum EmailType {
+    NewPost,
+    Newsletter,
+}
+
+impl Display for EmailType {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        match self {
+            EmailType::NewPost => write!(f, "new_post"),
+            EmailType::Newsletter => write!(f, "newsletter"),
+        }
+    }
+}
+
 #[tracing::instrument(skip_all)]
 pub async fn enqueue_delivery_tasks(
     transaction: &mut Transaction<'static, Postgres>,
     newsletter_issue_id: Uuid,
+    kind: EmailType,
 ) -> Result<(), sqlx::Error> {
     let query = sqlx::query!(
         r#"
         INSERT INTO issue_delivery_queue (
             newsletter_issue_id,
             subscriber_email,
-            unsubscribe_token
+            unsubscribe_token,
+            kind
         )
-        SELECT $1, email, unsubscribe_token
+        SELECT $1, email, unsubscribe_token, $2
         FROM subscriptions
         WHERE status = 'confirmed'
         "#,
         newsletter_issue_id,
+        kind.to_string()
     );
     transaction.execute(query).await?;
     Ok(())
@@ -102,7 +121,7 @@ pub async fn publish_newsletter(
         .await
         .context("Failed to store newsletter issue details.")?;
 
-    enqueue_delivery_tasks(&mut transaction, issue_id)
+    enqueue_delivery_tasks(&mut transaction, issue_id, EmailType::Newsletter)
         .await
         .context("Failed to enqueue delivery tasks.")?;
 

src/routes/admin/posts.rs

@@ -1,7 +1,7 @@
 use crate::{
     authentication::AuthenticatedUser,
     idempotency::{IdempotencyKey, save_response, try_processing},
-    routes::{AdminError, AppError, enqueue_delivery_tasks, insert_newsletter_issue},
+    routes::{AdminError, AppError, EmailType, enqueue_delivery_tasks, insert_newsletter_issue},
     startup::AppState,
     templates::{MessageTemplate, NewPostEmailTemplate},
 };
@@ -63,7 +63,7 @@ pub async fn create_post(
         .await
         .context("Failed to create newsletter.")?;
 
-    enqueue_delivery_tasks(&mut transaction, newsletter_uuid)
+    enqueue_delivery_tasks(&mut transaction, newsletter_uuid, EmailType::NewPost)
         .await
         .context("Failed to enqueue delivery tasks.")?;
 

src/routes/posts.rs

@@ -8,7 +8,7 @@ use anyhow::Context;
 use askama::Template;
 use axum::{
     extract::{Path, Query, State},
-    response::{Html, IntoResponse, Response},
+    response::{Html, IntoResponse, Redirect, Response},
 };
 use sqlx::PgPool;
 use uuid::Uuid;
@@ -62,12 +62,29 @@ async fn get_posts_table_size(connection_pool: &PgPool) -> Result<i64, sqlx::Err
         .map(|r| r.count.unwrap())
 }
 
+#[derive(serde::Deserialize)]
+pub struct PostParams {
+    origin: Option<Uuid>,
+}
+
 pub async fn see_post(
     State(AppState {
         connection_pool, ..
     }): State<AppState>,
     Path(post_id): Path<Uuid>,
+    Query(PostParams { origin }): Query<PostParams>,
 ) -> Result<Response, AppError> {
+    if let Some(origin) = origin {
+        sqlx::query!(
+            "UPDATE notifications_delivered SET opened = TRUE WHERE email_id = $1",
+            origin,
+        )
+        .execute(&connection_pool)
+        .await
+        .context("Failed to mark email as opened.")
+        .map_err(AppError::unexpected_page)?;
+        return Ok(Redirect::to(&format!("/posts/{}", post_id)).into_response());
+    }
     let post = get_post_data(&connection_pool, post_id)
         .await
         .context(format!("Failed to fetch post #{}", post_id))

src/templates.rs

@@ -1,4 +1,4 @@
-use crate::domain::PostEntry;
+use crate::{domain::PostEntry, routes::DashboardStats};
 use askama::Template;
 use uuid::Uuid;
 
@@ -24,6 +24,7 @@ pub struct DashboardTemplate {
     pub username: String,
     pub idempotency_key_1: String,
     pub idempotency_key_2: String,
+    pub stats: DashboardStats,
 }
 
 #[derive(Template)]
@@ -102,7 +103,7 @@ I just published a new post that I think you'll find interesting:
 
 "{}"
 
-Read the full post: {}/posts/{}
+Read the full post: {}/posts/{}?origin=EMAIL_ID
 
 This post covers practical insights and real-world examples that I hope will be valuable for your backend development journey.