Notification registration API baseline and email confirmation (#3)

Author: zupzup

Cargo.lock

@@ -21,9 +21,16 @@ tracing = "0.1.41"
 tracing-subscriber = "0.3.19"
 tower-http = { version = "0.6.6", features = ["cors"] }
 clap = { version = "4.5.39", features = ["derive", "env"] }
-chrono = "0.4.41"
+chrono = "0.4"
 serde_json = "1"
 serde = { version = "1", features = ["derive"] }
 async-trait = "0.1"
-tokio-postgres = "0.7.13"
-deadpool-postgres = "0.14.1"
+tokio-postgres = { version = "0.7", features = ["with-chrono-0_4"] }
+deadpool-postgres = "0.14"
+hex = { version = "0.4" }
+rand = { version = "0.8" }
+url = { version = "2.5" }
+email_address = "0.2"
+bitflags = "2.9"
+reqwest = { version = "0.12", default-features = false, features = ["default", "json"] }
+base64 = "0.22"

Cargo.toml

@@ -1,9 +1,9 @@
 use std::str::FromStr;
 
+use crate::db::PostgresStore;
 use async_trait::async_trait;
-use deadpool_postgres::{Manager, ManagerConfig, Pool, RecyclingMethod};
 use nostr::hashes::sha256::Hash as Sha256Hash;
-use tokio_postgres::{NoTls, Row};
+use tokio_postgres::Row;
 
 use super::File;
 
@@ -13,40 +13,8 @@ pub trait FileStoreApi: Send + Sync {
     async fn insert(&self, file: File) -> Result<(), anyhow::Error>;
 }
 
-pub struct PostgresFileStore {
-    pool: Pool,
-}
-
-impl PostgresFileStore {
-    pub async fn new(conn_str: &str) -> Result<Self, anyhow::Error> {
-        let cfg: tokio_postgres::Config = conn_str.parse()?;
-        let mgr_config = ManagerConfig {
-            recycling_method: RecyclingMethod::Fast,
-        };
-        let pool = Pool::builder(Manager::from_config(cfg, NoTls, mgr_config))
-            .max_size(16)
-            .build()?;
-
-        Ok(Self { pool })
-    }
-
-    /// Creates the table, if it doesn't exist yet
-    pub async fn init(&self) -> Result<(), anyhow::Error> {
-        let qry = r#"
-            CREATE TABLE IF NOT EXISTS files (
-                hash CHAR(64) PRIMARY KEY,
-                data BYTEA NOT NULL,
-                size INTEGER NOT NULL
-            )
-        "#;
-
-        self.pool.get().await?.execute(qry, &[]).await?;
-        Ok(())
-    }
-}
-
 #[async_trait]
-impl FileStoreApi for PostgresFileStore {
+impl FileStoreApi for PostgresStore {
     async fn get(&self, hash: &Sha256Hash) -> Result<Option<File>, anyhow::Error> {
         let row = self
             .pool

src/blossom/file_store.rs

@@ -0,0 +1,68 @@
+use deadpool_postgres::{Manager, ManagerConfig, Pool, RecyclingMethod};
+use tokio_postgres::NoTls;
+
+pub struct PostgresStore {
+    pub pool: Pool,
+}
+
+impl PostgresStore {
+    pub async fn new(conn_str: &str) -> Result<Self, anyhow::Error> {
+        let cfg: tokio_postgres::Config = conn_str.parse()?;
+        let mgr_config = ManagerConfig {
+            recycling_method: RecyclingMethod::Fast,
+        };
+        let pool = Pool::builder(Manager::from_config(cfg, NoTls, mgr_config))
+            .max_size(16)
+            .build()?;
+
+        Ok(Self { pool })
+    }
+
+    /// Creates the tables, if they don't exist yet
+    pub async fn init(&self) -> Result<(), anyhow::Error> {
+        // File Store
+        let qry = r#"
+            CREATE TABLE IF NOT EXISTS files (
+                hash CHAR(64) PRIMARY KEY,
+                data BYTEA NOT NULL,
+                size INTEGER NOT NULL
+            )
+        "#;
+        self.pool.get().await?.execute(qry, &[]).await?;
+
+        // Notification Store
+        let qry = r#"
+            CREATE TABLE IF NOT EXISTS notif_challenges (
+                npub TEXT PRIMARY KEY,
+                challenge TEXT NOT NULL,
+                created_at TIMESTAMPTZ DEFAULT (NOW() AT TIME ZONE 'UTC')
+            )
+        "#;
+        self.pool.get().await?.execute(qry, &[]).await?;
+
+        let qry = r#"
+            CREATE TABLE IF NOT EXISTS notif_email_verification (
+                npub TEXT PRIMARY KEY,
+                email TEXT NOT NULL,
+                confirmed BOOLEAN DEFAULT FALSE,
+                token TEXT,
+                sent_at TIMESTAMPTZ DEFAULT (NOW() AT TIME ZONE 'UTC')
+            )
+        "#;
+        self.pool.get().await?.execute(qry, &[]).await?;
+
+        let qry = r#"
+            CREATE TABLE IF NOT EXISTS notif_email_preferences (
+                npub TEXT PRIMARY KEY,
+                enabled BOOLEAN DEFAULT FALSE,
+                token TEXT NOT NULL,
+                email TEXT NOT NULL,
+                email_confirmed BOOLEAN DEFAULT FALSE,
+                ebill_url TEXT NOT NULL,
+                flags BIGINT NOT NULL
+            )
+        "#;
+        self.pool.get().await?.execute(qry, &[]).await?;
+        Ok(())
+    }
+}

src/db/mod.rs

@@ -1,5 +1,8 @@
 mod blossom;
+mod db;
+mod notification;
 mod relay;
+mod util;
 
 use std::{net::SocketAddr, sync::Arc};
 
@@ -8,7 +11,7 @@ use axum::{
     extract::{ConnectInfo, State},
     http::{StatusCode, Uri},
     response::IntoResponse,
-    routing::{any, delete, get, head, put},
+    routing::{any, delete, get, head, post, put},
     serve, Router,
 };
 use axum_raw_websocket::RawSocketUpgrade;
@@ -20,6 +23,14 @@ use relay::RelayConfig;
 use tower_http::cors::{Any, CorsLayer};
 use tracing::{error, info};
 
+use crate::notification::{
+    email::{
+        mailjet::{MailjetConfig, MailjetService},
+        EmailService,
+    },
+    notification_store::NotificationStoreApi,
+};
+
 #[tokio::main]
 async fn main() -> Result<()> {
     tracing_subscriber::fmt::init();
@@ -43,6 +54,12 @@ async fn main() -> Result<()> {
         .route("/{hash}", get(blossom::handle_get_file))
         .route("/{hash}", head(blossom::handle_get_file_head))
         .route("/", delete(blossom::handle_delete))
+        .route("/notifications/v1/start", post(notification::start))
+        .route("/notifications/v1/register", post(notification::register))
+        .route(
+            "/notifications/confirm_email",
+            get(notification::confirm_email),
+        )
         .route("/", any(websocket_handler))
         .fallback(handle_404)
         .with_state(app_state)
@@ -80,26 +97,38 @@ async fn handle_404(uri: Uri) -> impl IntoResponse {
 #[derive(Clone)]
 struct AppConfig {
     pub host_url: Url,
+    pub email_from_address: String,
 }
 
 #[derive(Clone)]
 struct AppState {
     pub relay: LocalRelay,
     pub cfg: AppConfig,
     pub file_store: Arc<dyn FileStoreApi>,
+    pub notification_store: Arc<dyn NotificationStoreApi>,
+    pub email_service: Arc<dyn EmailService>,
 }
 
 impl AppState {
     pub async fn new(config: &RelayConfig) -> Result<Self> {
-        let file_store =
-            blossom::file_store::PostgresFileStore::new(&config.db_connection_string()).await?;
-        file_store.init().await?;
+        let db = db::PostgresStore::new(&config.db_connection_string()).await?;
+        db.init().await?;
+        let store = Arc::new(db);
+
+        let email_service = MailjetService::new(&MailjetConfig {
+            api_key: config.email_api_key.clone(),
+            api_secret_key: config.email_api_secret_key.clone(),
+            url: config.email_url.clone(),
+        });
         Ok(Self {
             relay: relay::init(config).await?,
             cfg: AppConfig {
                 host_url: config.host_url.clone(),
+                email_from_address: config.email_from_address.clone(),
             },
-            file_store: Arc::new(file_store),
+            file_store: store.clone(),
+            notification_store: store,
+            email_service: Arc::new(email_service),
         })
     }
 }

src/main.rs

@@ -0,0 +1,119 @@
+use crate::notification::email::{EmailMessage, EmailService};
+use anyhow::anyhow;
+use async_trait::async_trait;
+use serde::{Deserialize, Serialize};
+use tracing::error;
+
+#[derive(Debug, Clone)]
+pub struct MailjetConfig {
+    pub api_key: String,
+    pub api_secret_key: String,
+    pub url: url::Url,
+}
+
+pub struct MailjetService {
+    config: MailjetConfig,
+    client: reqwest::Client,
+}
+
+impl MailjetService {
+    pub fn new(config: &MailjetConfig) -> Self {
+        let client = reqwest::Client::new();
+        Self {
+            config: config.to_owned(),
+            client,
+        }
+    }
+}
+
+#[derive(Debug, Clone, Serialize)]
+struct MailjetReq {
+    #[serde(rename = "Messages")]
+    pub messages: Vec<MailjetMessage>,
+}
+
+#[derive(Debug, Clone, Deserialize)]
+struct MailjetResp {
+    #[serde(rename = "Messages")]
+    pub messages: Vec<MailjetRespMessage>,
+}
+
+#[derive(Debug, Clone, Serialize)]
+struct MailjetMessage {
+    #[serde(rename = "From")]
+    pub from: MailjetFrom,
+    #[serde(rename = "To")]
+    pub to: Vec<MailjetTo>,
+    #[serde(rename = "Subject")]
+    pub subject: String,
+    #[serde(rename = "HTMLPart")]
+    pub html_part: String,
+}
+
+impl From<EmailMessage> for MailjetMessage {
+    fn from(value: EmailMessage) -> Self {
+        Self {
+            from: MailjetFrom { email: value.from },
+            to: vec![MailjetTo { email: value.to }],
+            subject: value.subject,
+            html_part: value.body,
+        }
+    }
+}
+
+#[derive(Debug, Clone, Serialize)]
+struct MailjetFrom {
+    #[serde(rename = "Email")]
+    pub email: String,
+}
+
+#[derive(Debug, Clone, Serialize)]
+struct MailjetTo {
+    #[serde(rename = "Email")]
+    pub email: String,
+}
+
+#[derive(Debug, Clone, Deserialize)]
+struct MailjetRespMessage {
+    #[serde(rename = "Status")]
+    pub status: String,
+}
+
+#[async_trait]
+impl EmailService for MailjetService {
+    async fn send(&self, msg: super::EmailMessage) -> Result<(), anyhow::Error> {
+        let mailjet_msg = MailjetReq {
+            messages: vec![MailjetMessage::from(msg)],
+        };
+
+        let url = self.config.url.join("/v3.1/send").expect("mailjet path");
+        let request = self.client.post(url).json(&mailjet_msg).basic_auth(
+            self.config.api_key.clone(),
+            Some(self.config.api_secret_key.clone()),
+        );
+        let res = request.send().await.map_err(|e| {
+            error!("Failed to send email: {e}");
+            anyhow!("Failed to send email")
+        })?;
+
+        let resp: MailjetResp = res.json().await.map_err(|e| {
+            error!("Failed to parse email response: {e}");
+            anyhow!("Failed to parse email response")
+        })?;
+
+        match resp.messages.first() {
+            Some(msg) => {
+                if msg.status != "success" {
+                    error!("Invalid email sending response: {}", &msg.status);
+                    Err(anyhow!("Invalid email sending response: {}", &msg.status))
+                } else {
+                    Ok(())
+                }
+            }
+            None => {
+                error!("Invalid email response - got no status");
+                Err(anyhow!("Invalid email response - got no status"))
+            }
+        }
+    }
+}

src/notification/email/mailjet.rs

@@ -0,0 +1,42 @@
+use async_trait::async_trait;
+
+pub mod mailjet;
+
+#[async_trait]
+pub trait EmailService: Send + Sync {
+    async fn send(&self, msg: EmailMessage) -> Result<(), anyhow::Error>;
+}
+
+/// A simple email message. We can add more features (like html, multi recipient, etc.) later.
+#[derive(Debug, Clone)]
+pub struct EmailMessage {
+    pub from: String,
+    pub to: String,
+    pub subject: String,
+    pub body: String,
+}
+
+pub fn build_email_confirmation_message(
+    host_url: &url::Url,
+    from: &str,
+    to: &str,
+    token: &str,
+) -> EmailMessage {
+    // build email confirmation link
+    let mut link = host_url
+        .join("/notifications/confirm_email")
+        .expect("email confirmation mail");
+    link.set_query(Some(&format!("token={token}")));
+
+    // build template
+    let body = format!(
+        "<html><head></head><body><a href=\"{link}\">Click here to confirm</a><br /><br />This link is valid for 1 day.</body></html>"
+    );
+
+    EmailMessage {
+        from: from.to_owned(),
+        to: to.to_owned(),
+        subject: "Confirm your E-Mail".to_owned(),
+        body: body.to_owned(),
+    }
+}