diff --git a/.env.example b/.env.example index f5305c5..9df8af7 100644 --- a/.env.example +++ b/.env.example @@ -13,6 +13,23 @@ AWS_SECRET_ACCESS_KEY=minioadmin PDS_HOSTNAME=localhost:3000 PLC_URL=plc.directory +# Notification Service Configuration +# At least one notification channel should be configured for user notifications to work. +# Email notifications (via sendmail/msmtp) +# MAIL_FROM_ADDRESS=noreply@example.com +# MAIL_FROM_NAME=My PDS +# SENDMAIL_PATH=/usr/sbin/sendmail + +# Discord notifications (not yet implemented) +# DISCORD_BOT_TOKEN=your-bot-token + +# Telegram notifications (not yet implemented) +# TELEGRAM_BOT_TOKEN=your-bot-token + +# Signal notifications (not yet implemented) +# SIGNAL_CLI_PATH=/usr/local/bin/signal-cli +# SIGNAL_PHONE_NUMBER=+1234567890 + CARGO_MOMMYS_LITTLE=mister CARGO_MOMMYS_PRONOUNS=his CARGO_MOMMYS_ROLES=daddy diff --git a/.sqlx/query-2c6cb8f15fe71cb5f38ffd7f5085b60bc852c4f1042c95a76fce773efd369511.json b/.sqlx/query-2c6cb8f15fe71cb5f38ffd7f5085b60bc852c4f1042c95a76fce773efd369511.json new file mode 100644 index 0000000..872ce2a --- /dev/null +++ b/.sqlx/query-2c6cb8f15fe71cb5f38ffd7f5085b60bc852c4f1042c95a76fce773efd369511.json @@ -0,0 +1,15 @@ +{ + "db_name": "PostgreSQL", + "query": "\n UPDATE notification_queue\n SET\n status = CASE\n WHEN attempts + 1 >= max_attempts THEN 'failed'::notification_status\n ELSE 'pending'::notification_status\n END,\n attempts = attempts + 1,\n last_error = $2,\n updated_at = NOW(),\n scheduled_for = NOW() + (INTERVAL '1 minute' * (attempts + 1))\n WHERE id = $1\n ", + "describe": { + "columns": [], + "parameters": { + "Left": [ + "Uuid", + "Text" + ] + }, + "nullable": [] + }, + "hash": "2c6cb8f15fe71cb5f38ffd7f5085b60bc852c4f1042c95a76fce773efd369511" +} diff --git a/.sqlx/query-303777d97e6ed344f8c699eae37b7b0c241c734a5b7726019c2a59ae277caee6.json b/.sqlx/query-303777d97e6ed344f8c699eae37b7b0c241c734a5b7726019c2a59ae277caee6.json new file mode 100644 index 0000000..967a3a5 --- /dev/null +++ b/.sqlx/query-303777d97e6ed344f8c699eae37b7b0c241c734a5b7726019c2a59ae277caee6.json @@ -0,0 +1,53 @@ +{ + "db_name": "PostgreSQL", + "query": "\n INSERT INTO notification_queue\n (user_id, channel, notification_type, recipient, subject, body, metadata)\n VALUES ($1, $2, $3, $4, $5, $6, $7)\n RETURNING id\n ", + "describe": { + "columns": [ + { + "ordinal": 0, + "name": "id", + "type_info": "Uuid" + } + ], + "parameters": { + "Left": [ + "Uuid", + { + "Custom": { + "name": "notification_channel", + "kind": { + "Enum": [ + "email", + "discord", + "telegram", + "signal" + ] + } + } + }, + { + "Custom": { + "name": "notification_type", + "kind": { + "Enum": [ + "welcome", + "email_verification", + "password_reset", + "email_update", + "account_deletion" + ] + } + } + }, + "Text", + "Text", + "Text", + "Jsonb" + ] + }, + "nullable": [ + false + ] + }, + "hash": "303777d97e6ed344f8c699eae37b7b0c241c734a5b7726019c2a59ae277caee6" +} diff --git a/.sqlx/query-344c851d3f1b026e8632aa2f04052dcbc957b7077c856da6a1a256ec2fe85ad3.json b/.sqlx/query-344c851d3f1b026e8632aa2f04052dcbc957b7077c856da6a1a256ec2fe85ad3.json new file mode 100644 index 0000000..7abae00 --- /dev/null +++ b/.sqlx/query-344c851d3f1b026e8632aa2f04052dcbc957b7077c856da6a1a256ec2fe85ad3.json @@ -0,0 +1,14 @@ +{ + "db_name": "PostgreSQL", + "query": "\n UPDATE notification_queue\n SET status = 'sent', processed_at = NOW(), updated_at = NOW()\n WHERE id = $1\n ", + "describe": { + "columns": [], + "parameters": { + "Left": [ + "Uuid" + ] + }, + "nullable": [] + }, + "hash": "344c851d3f1b026e8632aa2f04052dcbc957b7077c856da6a1a256ec2fe85ad3" +} diff --git a/.sqlx/query-55c4e13e5ff23aaa71c3ab417891a5f56542571ba3f15c6d9dae153405bc4275.json b/.sqlx/query-55c4e13e5ff23aaa71c3ab417891a5f56542571ba3f15c6d9dae153405bc4275.json new file mode 100644 index 0000000..8284690 --- /dev/null +++ b/.sqlx/query-55c4e13e5ff23aaa71c3ab417891a5f56542571ba3f15c6d9dae153405bc4275.json @@ -0,0 +1,46 @@ +{ + "db_name": "PostgreSQL", + "query": "\n SELECT s.did, u.id as user_id, u.email, u.handle, k.key_bytes\n FROM sessions s\n JOIN users u ON s.did = u.did\n JOIN user_keys k ON u.id = k.user_id\n WHERE s.access_jwt = $1\n ", + "describe": { + "columns": [ + { + "ordinal": 0, + "name": "did", + "type_info": "Text" + }, + { + "ordinal": 1, + "name": "user_id", + "type_info": "Uuid" + }, + { + "ordinal": 2, + "name": "email", + "type_info": "Text" + }, + { + "ordinal": 3, + "name": "handle", + "type_info": "Text" + }, + { + "ordinal": 4, + "name": "key_bytes", + "type_info": "Bytea" + } + ], + "parameters": { + "Left": [ + "Text" + ] + }, + "nullable": [ + false, + false, + false, + false, + false + ] + }, + "hash": "55c4e13e5ff23aaa71c3ab417891a5f56542571ba3f15c6d9dae153405bc4275" +} diff --git a/.sqlx/query-5d49bbf0307a0c642b0174d641de748fa648c97f8109255120e969c957ff95bf.json b/.sqlx/query-5d49bbf0307a0c642b0174d641de748fa648c97f8109255120e969c957ff95bf.json new file mode 100644 index 0000000..d57b1a5 --- /dev/null +++ b/.sqlx/query-5d49bbf0307a0c642b0174d641de748fa648c97f8109255120e969c957ff95bf.json @@ -0,0 +1,53 @@ +{ + "db_name": "PostgreSQL", + "query": "\n INSERT INTO notification_queue\n (user_id, channel, notification_type, recipient, subject, body, metadata)\n VALUES ($1, $2, $3, $4, $5, $6, $7)\n RETURNING id\n ", + "describe": { + "columns": [ + { + "ordinal": 0, + "name": "id", + "type_info": "Uuid" + } + ], + "parameters": { + "Left": [ + "Uuid", + { + "Custom": { + "name": "notification_channel", + "kind": { + "Enum": [ + "email", + "discord", + "telegram", + "signal" + ] + } + } + }, + { + "Custom": { + "name": "notification_type", + "kind": { + "Enum": [ + "welcome", + "email_verification", + "password_reset", + "email_update", + "account_deletion" + ] + } + } + }, + "Text", + "Text", + "Text", + "Jsonb" + ] + }, + "nullable": [ + false + ] + }, + "hash": "5d49bbf0307a0c642b0174d641de748fa648c97f8109255120e969c957ff95bf" +} diff --git a/.sqlx/query-cb6f48aaba124c79308d20e66c23adb44d1196296b7f93fad19b2d17548ed3de.json b/.sqlx/query-cb6f48aaba124c79308d20e66c23adb44d1196296b7f93fad19b2d17548ed3de.json new file mode 100644 index 0000000..9072434 --- /dev/null +++ b/.sqlx/query-cb6f48aaba124c79308d20e66c23adb44d1196296b7f93fad19b2d17548ed3de.json @@ -0,0 +1,150 @@ +{ + "db_name": "PostgreSQL", + "query": "\n UPDATE notification_queue\n SET status = 'processing', updated_at = NOW()\n WHERE id IN (\n SELECT id FROM notification_queue\n WHERE status = 'pending'\n AND scheduled_for <= $1\n AND attempts < max_attempts\n ORDER BY scheduled_for ASC\n LIMIT $2\n FOR UPDATE SKIP LOCKED\n )\n RETURNING\n id, user_id,\n channel as \"channel: NotificationChannel\",\n notification_type as \"notification_type: super::types::NotificationType\",\n status as \"status: NotificationStatus\",\n recipient, subject, body, metadata,\n attempts, max_attempts, last_error,\n created_at, updated_at, scheduled_for, processed_at\n ", + "describe": { + "columns": [ + { + "ordinal": 0, + "name": "id", + "type_info": "Uuid" + }, + { + "ordinal": 1, + "name": "user_id", + "type_info": "Uuid" + }, + { + "ordinal": 2, + "name": "channel: NotificationChannel", + "type_info": { + "Custom": { + "name": "notification_channel", + "kind": { + "Enum": [ + "email", + "discord", + "telegram", + "signal" + ] + } + } + } + }, + { + "ordinal": 3, + "name": "notification_type: super::types::NotificationType", + "type_info": { + "Custom": { + "name": "notification_type", + "kind": { + "Enum": [ + "welcome", + "email_verification", + "password_reset", + "email_update", + "account_deletion" + ] + } + } + } + }, + { + "ordinal": 4, + "name": "status: NotificationStatus", + "type_info": { + "Custom": { + "name": "notification_status", + "kind": { + "Enum": [ + "pending", + "processing", + "sent", + "failed" + ] + } + } + } + }, + { + "ordinal": 5, + "name": "recipient", + "type_info": "Text" + }, + { + "ordinal": 6, + "name": "subject", + "type_info": "Text" + }, + { + "ordinal": 7, + "name": "body", + "type_info": "Text" + }, + { + "ordinal": 8, + "name": "metadata", + "type_info": "Jsonb" + }, + { + "ordinal": 9, + "name": "attempts", + "type_info": "Int4" + }, + { + "ordinal": 10, + "name": "max_attempts", + "type_info": "Int4" + }, + { + "ordinal": 11, + "name": "last_error", + "type_info": "Text" + }, + { + "ordinal": 12, + "name": "created_at", + "type_info": "Timestamptz" + }, + { + "ordinal": 13, + "name": "updated_at", + "type_info": "Timestamptz" + }, + { + "ordinal": 14, + "name": "scheduled_for", + "type_info": "Timestamptz" + }, + { + "ordinal": 15, + "name": "processed_at", + "type_info": "Timestamptz" + } + ], + "parameters": { + "Left": [ + "Timestamptz", + "Int8" + ] + }, + "nullable": [ + false, + false, + false, + false, + false, + false, + true, + false, + true, + false, + false, + true, + false, + false, + false, + true + ] + }, + "hash": "cb6f48aaba124c79308d20e66c23adb44d1196296b7f93fad19b2d17548ed3de" +} diff --git a/Cargo.toml b/Cargo.toml index e831c9f..acb2320 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -30,7 +30,7 @@ serde_json = "1.0.145" sha2 = "0.10.9" sqlx = { version = "0.8.6", features = ["runtime-tokio-rustls", "postgres", "uuid", "chrono", "json"] } thiserror = "2.0.17" -tokio = { version = "1.48.0", features = ["macros", "rt-multi-thread", "time"] } +tokio = { version = "1.48.0", features = ["macros", "rt-multi-thread", "time", "signal", "process"] } tracing = "0.1.43" tracing-subscriber = "0.3.22" uuid = { version = "1.19.0", features = ["v4", "fast-rng"] } diff --git a/TODO.md b/TODO.md index 06f2375..51a8b48 100644 --- a/TODO.md +++ b/TODO.md @@ -126,10 +126,16 @@ Lewis' corrected big boy todofile - [ ] Implement caching layer for DID resolution (Redis or in-memory). - [ ] Handle cache invalidation/expiry. - [ ] Background Jobs - - [ ] Implement background queue for async tasks (crawler notifications, discord/telegram 2FA sending instead of email). - [ ] Implement `Crawlers` service (debounce notifications to relays). -- [ ] Mailer equivalent - - [ ] Implement code/notification sending service as a replacement for the mailer because there's no way I'm starting with email. :D +- [x] Notification Service + - [x] Queue-based notification system with database table + - [x] Background worker polling for pending notifications + - [x] Extensible sender trait for multiple channels + - [x] Email sender via OS sendmail/msmtp + - [ ] Discord bot sender + - [ ] Telegram bot sender + - [ ] Signal bot sender + - [x] Helper functions for common notification types (welcome, password reset, email verification, etc.) - [ ] Image Processing - [ ] Implement image resize/formatting pipeline (for blob uploads). - [ ] IPLD & MST diff --git a/migrations/202512212000_notification_queue.sql b/migrations/202512212000_notification_queue.sql new file mode 100644 index 0000000..5315185 --- /dev/null +++ b/migrations/202512212000_notification_queue.sql @@ -0,0 +1,36 @@ +CREATE TYPE notification_channel AS ENUM ('email', 'discord', 'telegram', 'signal'); +CREATE TYPE notification_status AS ENUM ('pending', 'processing', 'sent', 'failed'); +CREATE TYPE notification_type AS ENUM ( + 'welcome', + 'email_verification', + 'password_reset', + 'email_update', + 'account_deletion' +); + +CREATE TABLE IF NOT EXISTS notification_queue ( + id UUID PRIMARY KEY DEFAULT gen_random_uuid(), + user_id UUID NOT NULL REFERENCES users(id) ON DELETE CASCADE, + channel notification_channel NOT NULL DEFAULT 'email', + notification_type notification_type NOT NULL, + status notification_status NOT NULL DEFAULT 'pending', + recipient TEXT NOT NULL, + subject TEXT, + body TEXT NOT NULL, + metadata JSONB, + attempts INT NOT NULL DEFAULT 0, + max_attempts INT NOT NULL DEFAULT 3, + last_error TEXT, + created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(), + updated_at TIMESTAMPTZ NOT NULL DEFAULT NOW(), + scheduled_for TIMESTAMPTZ NOT NULL DEFAULT NOW(), + processed_at TIMESTAMPTZ +); + +CREATE INDEX idx_notification_queue_status_scheduled + ON notification_queue(status, scheduled_for) + WHERE status = 'pending'; + +CREATE INDEX idx_notification_queue_user_id ON notification_queue(user_id); + +ALTER TABLE users ADD COLUMN IF NOT EXISTS preferred_notification_channel notification_channel NOT NULL DEFAULT 'email'; diff --git a/src/api/identity/account.rs b/src/api/identity/account.rs index 9339f2d..88e3445 100644 --- a/src/api/identity/account.rs +++ b/src/api/identity/account.rs @@ -14,7 +14,7 @@ use rand::rngs::OsRng; use serde::{Deserialize, Serialize}; use serde_json::json; use std::sync::Arc; -use tracing::{error, info}; +use tracing::{error, info, warn}; #[derive(Deserialize)] pub struct CreateAccountInput { @@ -332,6 +332,19 @@ pub async fn create_account( .into_response(); } + let hostname = std::env::var("PDS_HOSTNAME").unwrap_or_else(|_| "localhost".to_string()); + if let Err(e) = crate::notifications::enqueue_welcome_email( + &state.db, + user_id, + &input.email, + &input.handle, + &hostname, + ) + .await + { + warn!("Failed to enqueue welcome email: {:?}", e); + } + ( StatusCode::OK, Json(CreateAccountOutput { diff --git a/src/api/server/session.rs b/src/api/server/session.rs index 6dc68d1..e86669c 100644 --- a/src/api/server/session.rs +++ b/src/api/server/session.rs @@ -363,7 +363,7 @@ pub async fn request_account_delete( let session = sqlx::query!( r#" - SELECT s.did, k.key_bytes + SELECT s.did, u.id as user_id, u.email, u.handle, k.key_bytes FROM sessions s JOIN users u ON s.did = u.did JOIN user_keys k ON u.id = k.user_id @@ -374,8 +374,8 @@ pub async fn request_account_delete( .fetch_optional(&state.db) .await; - let (did, key_bytes) = match session { - Ok(Some(row)) => (row.did, row.key_bytes), + let (did, user_id, email, handle, key_bytes) = match session { + Ok(Some(row)) => (row.did, row.user_id, row.email, row.handle, row.key_bytes), Ok(None) => { return ( StatusCode::UNAUTHORIZED, @@ -422,8 +422,21 @@ pub async fn request_account_delete( .into_response(); } - // TODO: Send email or other notification - info!("Account deletion requested for user {}, token: {}", did, confirmation_token); + let hostname = std::env::var("PDS_HOSTNAME").unwrap_or_else(|_| "localhost".to_string()); + if let Err(e) = crate::notifications::enqueue_account_deletion( + &state.db, + user_id, + &email, + &handle, + &confirmation_token, + &hostname, + ) + .await + { + warn!("Failed to enqueue account deletion notification: {:?}", e); + } + + info!("Account deletion requested for user {}", did); (StatusCode::OK, Json(json!({}))).into_response() } diff --git a/src/lib.rs b/src/lib.rs index c3b12b8..d90506a 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -1,5 +1,6 @@ pub mod api; pub mod auth; +pub mod notifications; pub mod repo; pub mod state; pub mod storage; diff --git a/src/main.rs b/src/main.rs index 7db7e6d..aecb98a 100644 --- a/src/main.rs +++ b/src/main.rs @@ -1,6 +1,8 @@ +use bspds::notifications::{EmailSender, NotificationService}; use bspds::state::AppState; use std::net::SocketAddr; -use tracing::info; +use tokio::sync::watch; +use tracing::{info, warn}; #[tokio::main] async fn main() { @@ -20,12 +22,61 @@ async fn main() { .await .expect("Failed to run migrations"); - let state = AppState::new(pool).await; + let state = AppState::new(pool.clone()).await; + + let (shutdown_tx, shutdown_rx) = watch::channel(false); + + let mut notification_service = NotificationService::new(pool); + + if let Some(email_sender) = EmailSender::from_env() { + info!("Email notifications enabled"); + notification_service = notification_service.register_sender(email_sender); + } else { + warn!("Email notifications disabled (MAIL_FROM_ADDRESS not set)"); + } + + let notification_handle = tokio::spawn(notification_service.run(shutdown_rx)); let app = bspds::app(state); let addr = SocketAddr::from(([127, 0, 0, 1], 3000)); info!("listening on {}", addr); let listener = tokio::net::TcpListener::bind(addr).await.unwrap(); - axum::serve(listener, app).await.unwrap(); + + let server_result = axum::serve(listener, app) + .with_graceful_shutdown(shutdown_signal(shutdown_tx)) + .await; + + notification_handle.await.ok(); + + if let Err(e) = server_result { + tracing::error!("Server error: {}", e); + } +} + +async fn shutdown_signal(shutdown_tx: watch::Sender) { + let ctrl_c = async { + tokio::signal::ctrl_c() + .await + .expect("Failed to install Ctrl+C handler"); + }; + + #[cfg(unix)] + let terminate = async { + tokio::signal::unix::signal(tokio::signal::unix::SignalKind::terminate()) + .expect("Failed to install signal handler") + .recv() + .await; + }; + + #[cfg(not(unix))] + let terminate = std::future::pending::<()>(); + + tokio::select! { + _ = ctrl_c => {}, + _ = terminate => {}, + } + + info!("Shutdown signal received, stopping services..."); + shutdown_tx.send(true).ok(); } diff --git a/src/notifications/mod.rs b/src/notifications/mod.rs new file mode 100644 index 0000000..8d8bb47 --- /dev/null +++ b/src/notifications/mod.rs @@ -0,0 +1,12 @@ +mod sender; +mod service; +mod types; + +pub use sender::{EmailSender, NotificationSender}; +pub use service::{ + enqueue_account_deletion, enqueue_email_update, enqueue_email_verification, + enqueue_notification, enqueue_password_reset, enqueue_welcome_email, NotificationService, +}; +pub use types::{ + NewNotification, NotificationChannel, NotificationStatus, NotificationType, QueuedNotification, +}; diff --git a/src/notifications/sender.rs b/src/notifications/sender.rs new file mode 100644 index 0000000..888c8c5 --- /dev/null +++ b/src/notifications/sender.rs @@ -0,0 +1,98 @@ +use async_trait::async_trait; +use std::process::Stdio; +use tokio::io::AsyncWriteExt; +use tokio::process::Command; + +use super::types::{NotificationChannel, QueuedNotification}; + +#[async_trait] +pub trait NotificationSender: Send + Sync { + fn channel(&self) -> NotificationChannel; + async fn send(&self, notification: &QueuedNotification) -> Result<(), SendError>; +} + +#[derive(Debug, thiserror::Error)] +pub enum SendError { + #[error("Failed to spawn sendmail process: {0}")] + ProcessSpawn(#[from] std::io::Error), + + #[error("Sendmail exited with non-zero status: {0}")] + SendmailFailed(String), + + #[error("Channel not configured: {0:?}")] + NotConfigured(NotificationChannel), + + #[error("External service error: {0}")] + ExternalService(String), +} + +pub struct EmailSender { + from_address: String, + from_name: String, + sendmail_path: String, +} + +impl EmailSender { + pub fn new(from_address: String, from_name: String) -> Self { + Self { + from_address, + from_name, + sendmail_path: std::env::var("SENDMAIL_PATH").unwrap_or_else(|_| "/usr/sbin/sendmail".to_string()), + } + } + + pub fn from_env() -> Option { + let from_address = std::env::var("MAIL_FROM_ADDRESS").ok()?; + let from_name = std::env::var("MAIL_FROM_NAME").unwrap_or_else(|_| "BSPDS".to_string()); + Some(Self::new(from_address, from_name)) + } + + fn format_email(&self, notification: &QueuedNotification) -> String { + let subject = notification.subject.as_deref().unwrap_or("Notification"); + let from_header = if self.from_name.is_empty() { + self.from_address.clone() + } else { + format!("{} <{}>", self.from_name, self.from_address) + }; + + format!( + "From: {}\r\nTo: {}\r\nSubject: {}\r\nContent-Type: text/plain; charset=utf-8\r\nMIME-Version: 1.0\r\n\r\n{}", + from_header, + notification.recipient, + subject, + notification.body + ) + } +} + +#[async_trait] +impl NotificationSender for EmailSender { + fn channel(&self) -> NotificationChannel { + NotificationChannel::Email + } + + async fn send(&self, notification: &QueuedNotification) -> Result<(), SendError> { + let email_content = self.format_email(notification); + + let mut child = Command::new(&self.sendmail_path) + .arg("-t") + .arg("-oi") + .stdin(Stdio::piped()) + .stdout(Stdio::piped()) + .stderr(Stdio::piped()) + .spawn()?; + + if let Some(mut stdin) = child.stdin.take() { + stdin.write_all(email_content.as_bytes()).await?; + } + + let output = child.wait_with_output().await?; + + if !output.status.success() { + let stderr = String::from_utf8_lossy(&output.stderr); + return Err(SendError::SendmailFailed(stderr.to_string())); + } + + Ok(()) + } +} diff --git a/src/notifications/service.rs b/src/notifications/service.rs new file mode 100644 index 0000000..9508a49 --- /dev/null +++ b/src/notifications/service.rs @@ -0,0 +1,384 @@ +use std::collections::HashMap; +use std::sync::Arc; +use std::time::Duration; + +use chrono::Utc; +use sqlx::PgPool; +use tokio::sync::watch; +use tokio::time::interval; +use tracing::{debug, error, info, warn}; +use uuid::Uuid; + +use super::sender::{NotificationSender, SendError}; +use super::types::{NewNotification, NotificationChannel, NotificationStatus, QueuedNotification}; + +pub struct NotificationService { + db: PgPool, + senders: HashMap>, + poll_interval: Duration, + batch_size: i64, +} + +impl NotificationService { + pub fn new(db: PgPool) -> Self { + Self { + db, + senders: HashMap::new(), + poll_interval: Duration::from_secs(5), + batch_size: 10, + } + } + + pub fn with_poll_interval(mut self, interval: Duration) -> Self { + self.poll_interval = interval; + self + } + + pub fn with_batch_size(mut self, size: i64) -> Self { + self.batch_size = size; + self + } + + pub fn register_sender(mut self, sender: S) -> Self { + self.senders.insert(sender.channel(), Arc::new(sender)); + self + } + + pub async fn enqueue(&self, notification: NewNotification) -> Result { + let id = sqlx::query_scalar!( + r#" + INSERT INTO notification_queue + (user_id, channel, notification_type, recipient, subject, body, metadata) + VALUES ($1, $2, $3, $4, $5, $6, $7) + RETURNING id + "#, + notification.user_id, + notification.channel as NotificationChannel, + notification.notification_type as super::types::NotificationType, + notification.recipient, + notification.subject, + notification.body, + notification.metadata + ) + .fetch_one(&self.db) + .await?; + + debug!(notification_id = %id, "Notification enqueued"); + Ok(id) + } + + pub fn has_senders(&self) -> bool { + !self.senders.is_empty() + } + + pub async fn run(self, mut shutdown: watch::Receiver) { + if self.senders.is_empty() { + warn!("Notification service starting with no senders configured. Notifications will be queued but not delivered until senders are configured."); + } + + info!( + poll_interval_secs = self.poll_interval.as_secs(), + batch_size = self.batch_size, + channels = ?self.senders.keys().collect::>(), + "Starting notification service" + ); + + let mut ticker = interval(self.poll_interval); + + loop { + tokio::select! { + _ = ticker.tick() => { + if let Err(e) = self.process_batch().await { + error!(error = %e, "Failed to process notification batch"); + } + } + _ = shutdown.changed() => { + if *shutdown.borrow() { + info!("Notification service shutting down"); + break; + } + } + } + } + } + + async fn process_batch(&self) -> Result<(), sqlx::Error> { + let notifications = self.fetch_pending_notifications().await?; + + if notifications.is_empty() { + return Ok(()); + } + + debug!(count = notifications.len(), "Processing notification batch"); + + for notification in notifications { + self.process_notification(notification).await; + } + + Ok(()) + } + + async fn fetch_pending_notifications(&self) -> Result, sqlx::Error> { + let now = Utc::now(); + + sqlx::query_as!( + QueuedNotification, + r#" + UPDATE notification_queue + SET status = 'processing', updated_at = NOW() + WHERE id IN ( + SELECT id FROM notification_queue + WHERE status = 'pending' + AND scheduled_for <= $1 + AND attempts < max_attempts + ORDER BY scheduled_for ASC + LIMIT $2 + FOR UPDATE SKIP LOCKED + ) + RETURNING + id, user_id, + channel as "channel: NotificationChannel", + notification_type as "notification_type: super::types::NotificationType", + status as "status: NotificationStatus", + recipient, subject, body, metadata, + attempts, max_attempts, last_error, + created_at, updated_at, scheduled_for, processed_at + "#, + now, + self.batch_size + ) + .fetch_all(&self.db) + .await + } + + async fn process_notification(&self, notification: QueuedNotification) { + let notification_id = notification.id; + let channel = notification.channel; + + let result = match self.senders.get(&channel) { + Some(sender) => sender.send(¬ification).await, + None => { + warn!( + notification_id = %notification_id, + channel = ?channel, + "No sender registered for channel" + ); + Err(SendError::NotConfigured(channel)) + } + }; + + match result { + Ok(()) => { + debug!(notification_id = %notification_id, "Notification sent successfully"); + if let Err(e) = self.mark_sent(notification_id).await { + error!( + notification_id = %notification_id, + error = %e, + "Failed to mark notification as sent" + ); + } + } + Err(e) => { + let error_msg = e.to_string(); + warn!( + notification_id = %notification_id, + error = %error_msg, + "Failed to send notification" + ); + if let Err(db_err) = self.mark_failed(notification_id, &error_msg).await { + error!( + notification_id = %notification_id, + error = %db_err, + "Failed to mark notification as failed" + ); + } + } + } + } + + async fn mark_sent(&self, id: Uuid) -> Result<(), sqlx::Error> { + sqlx::query!( + r#" + UPDATE notification_queue + SET status = 'sent', processed_at = NOW(), updated_at = NOW() + WHERE id = $1 + "#, + id + ) + .execute(&self.db) + .await?; + Ok(()) + } + + async fn mark_failed(&self, id: Uuid, error: &str) -> Result<(), sqlx::Error> { + sqlx::query!( + r#" + UPDATE notification_queue + SET + status = CASE + WHEN attempts + 1 >= max_attempts THEN 'failed'::notification_status + ELSE 'pending'::notification_status + END, + attempts = attempts + 1, + last_error = $2, + updated_at = NOW(), + scheduled_for = NOW() + (INTERVAL '1 minute' * (attempts + 1)) + WHERE id = $1 + "#, + id, + error + ) + .execute(&self.db) + .await?; + Ok(()) + } +} + +pub async fn enqueue_notification(db: &PgPool, notification: NewNotification) -> Result { + sqlx::query_scalar!( + r#" + INSERT INTO notification_queue + (user_id, channel, notification_type, recipient, subject, body, metadata) + VALUES ($1, $2, $3, $4, $5, $6, $7) + RETURNING id + "#, + notification.user_id, + notification.channel as NotificationChannel, + notification.notification_type as super::types::NotificationType, + notification.recipient, + notification.subject, + notification.body, + notification.metadata + ) + .fetch_one(db) + .await +} + +pub async fn enqueue_welcome_email( + db: &PgPool, + user_id: Uuid, + email: &str, + handle: &str, + hostname: &str, +) -> Result { + let body = format!( + "Welcome to {}!\n\nYour handle is: @{}\n\nThank you for joining us.", + hostname, handle + ); + + enqueue_notification( + db, + NewNotification::email( + user_id, + super::types::NotificationType::Welcome, + email.to_string(), + format!("Welcome to {}", hostname), + body, + ), + ) + .await +} + +pub async fn enqueue_email_verification( + db: &PgPool, + user_id: Uuid, + email: &str, + handle: &str, + code: &str, + hostname: &str, +) -> Result { + let body = format!( + "Hello @{},\n\nYour email verification code is: {}\n\nThis code will expire in 10 minutes.\n\nIf you did not request this, please ignore this email.", + handle, code + ); + + enqueue_notification( + db, + NewNotification::email( + user_id, + super::types::NotificationType::EmailVerification, + email.to_string(), + format!("Verify your email - {}", hostname), + body, + ), + ) + .await +} + +pub async fn enqueue_password_reset( + db: &PgPool, + user_id: Uuid, + email: &str, + handle: &str, + code: &str, + hostname: &str, +) -> Result { + let body = format!( + "Hello @{},\n\nYour password reset code is: {}\n\nThis code will expire in 10 minutes.\n\nIf you did not request this, please ignore this email.", + handle, code + ); + + enqueue_notification( + db, + NewNotification::email( + user_id, + super::types::NotificationType::PasswordReset, + email.to_string(), + format!("Password Reset - {}", hostname), + body, + ), + ) + .await +} + +pub async fn enqueue_email_update( + db: &PgPool, + user_id: Uuid, + new_email: &str, + handle: &str, + code: &str, + hostname: &str, +) -> Result { + let body = format!( + "Hello @{},\n\nYour email update confirmation code is: {}\n\nThis code will expire in 10 minutes.\n\nIf you did not request this, please ignore this email.", + handle, code + ); + + enqueue_notification( + db, + NewNotification::email( + user_id, + super::types::NotificationType::EmailUpdate, + new_email.to_string(), + format!("Confirm your new email - {}", hostname), + body, + ), + ) + .await +} + +pub async fn enqueue_account_deletion( + db: &PgPool, + user_id: Uuid, + email: &str, + handle: &str, + code: &str, + hostname: &str, +) -> Result { + let body = format!( + "Hello @{},\n\nYour account deletion confirmation code is: {}\n\nThis code will expire in 10 minutes.\n\nIf you did not request this, please secure your account immediately.", + handle, code + ); + + enqueue_notification( + db, + NewNotification::email( + user_id, + super::types::NotificationType::AccountDeletion, + email.to_string(), + format!("Account Deletion Request - {}", hostname), + body, + ), + ) + .await +} diff --git a/src/notifications/types.rs b/src/notifications/types.rs new file mode 100644 index 0000000..23d0901 --- /dev/null +++ b/src/notifications/types.rs @@ -0,0 +1,82 @@ +use chrono::{DateTime, Utc}; +use serde::{Deserialize, Serialize}; +use sqlx::FromRow; +use uuid::Uuid; + +#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, sqlx::Type, Serialize, Deserialize)] +#[sqlx(type_name = "notification_channel", rename_all = "lowercase")] +pub enum NotificationChannel { + Email, + Discord, + Telegram, + Signal, +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq, sqlx::Type, Serialize, Deserialize)] +#[sqlx(type_name = "notification_status", rename_all = "lowercase")] +pub enum NotificationStatus { + Pending, + Processing, + Sent, + Failed, +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq, sqlx::Type, Serialize, Deserialize)] +#[sqlx(type_name = "notification_type", rename_all = "snake_case")] +pub enum NotificationType { + Welcome, + EmailVerification, + PasswordReset, + EmailUpdate, + AccountDeletion, +} + +#[derive(Debug, Clone, FromRow)] +pub struct QueuedNotification { + pub id: Uuid, + pub user_id: Uuid, + pub channel: NotificationChannel, + pub notification_type: NotificationType, + pub status: NotificationStatus, + pub recipient: String, + pub subject: Option, + pub body: String, + pub metadata: Option, + pub attempts: i32, + pub max_attempts: i32, + pub last_error: Option, + pub created_at: DateTime, + pub updated_at: DateTime, + pub scheduled_for: DateTime, + pub processed_at: Option>, +} + +pub struct NewNotification { + pub user_id: Uuid, + pub channel: NotificationChannel, + pub notification_type: NotificationType, + pub recipient: String, + pub subject: Option, + pub body: String, + pub metadata: Option, +} + +impl NewNotification { + pub fn email( + user_id: Uuid, + notification_type: NotificationType, + recipient: String, + subject: String, + body: String, + ) -> Self { + Self { + user_id, + channel: NotificationChannel::Email, + notification_type, + recipient, + subject: Some(subject), + body, + metadata: None, + } + } +} diff --git a/tests/notifications.rs b/tests/notifications.rs new file mode 100644 index 0000000..4ba3856 --- /dev/null +++ b/tests/notifications.rs @@ -0,0 +1,144 @@ +mod common; + +use bspds::notifications::{ + enqueue_notification, enqueue_welcome_email, NewNotification, NotificationChannel, + NotificationStatus, NotificationType, +}; +use sqlx::PgPool; + +async fn get_pool() -> PgPool { + let conn_str = common::get_db_connection_string().await; + sqlx::postgres::PgPoolOptions::new() + .max_connections(5) + .connect(&conn_str) + .await + .expect("Failed to connect to test database") +} + +#[tokio::test] +async fn test_enqueue_notification() { + let pool = get_pool().await; + + let (_, did) = common::create_account_and_login(&common::client()).await; + + let user_id: uuid::Uuid = sqlx::query_scalar!("SELECT id FROM users WHERE did = $1", did) + .fetch_one(&pool) + .await + .expect("User not found"); + + let notification = NewNotification::email( + user_id, + NotificationType::Welcome, + "test@example.com".to_string(), + "Test Subject".to_string(), + "Test body".to_string(), + ); + + let notification_id = enqueue_notification(&pool, notification) + .await + .expect("Failed to enqueue notification"); + + let row = sqlx::query!( + r#" + SELECT + id, user_id, recipient, subject, body, + channel as "channel: NotificationChannel", + notification_type as "notification_type: NotificationType", + status as "status: NotificationStatus" + FROM notification_queue + WHERE id = $1 + "#, + notification_id + ) + .fetch_one(&pool) + .await + .expect("Notification not found"); + + assert_eq!(row.user_id, user_id); + assert_eq!(row.recipient, "test@example.com"); + assert_eq!(row.subject.as_deref(), Some("Test Subject")); + assert_eq!(row.body, "Test body"); + assert_eq!(row.channel, NotificationChannel::Email); + assert_eq!(row.notification_type, NotificationType::Welcome); + assert_eq!(row.status, NotificationStatus::Pending); +} + +#[tokio::test] +async fn test_enqueue_welcome_email() { + let pool = get_pool().await; + + let (_, did) = common::create_account_and_login(&common::client()).await; + + let user_id: uuid::Uuid = sqlx::query_scalar!("SELECT id FROM users WHERE did = $1", did) + .fetch_one(&pool) + .await + .expect("User not found"); + + let notification_id = enqueue_welcome_email(&pool, user_id, "user@example.com", "testhandle", "example.com") + .await + .expect("Failed to enqueue welcome email"); + + let row = sqlx::query!( + r#" + SELECT + recipient, subject, body, + notification_type as "notification_type: NotificationType" + FROM notification_queue + WHERE id = $1 + "#, + notification_id + ) + .fetch_one(&pool) + .await + .expect("Notification not found"); + + assert_eq!(row.recipient, "user@example.com"); + assert_eq!(row.subject.as_deref(), Some("Welcome to example.com")); + assert!(row.body.contains("@testhandle")); + assert_eq!(row.notification_type, NotificationType::Welcome); +} + +#[tokio::test] +async fn test_notification_queue_status_index() { + let pool = get_pool().await; + + let (_, did) = common::create_account_and_login(&common::client()).await; + + let user_id: uuid::Uuid = sqlx::query_scalar!("SELECT id FROM users WHERE did = $1", did) + .fetch_one(&pool) + .await + .expect("User not found"); + + let initial_count: i64 = sqlx::query_scalar!( + "SELECT COUNT(*) FROM notification_queue WHERE status = 'pending' AND user_id = $1", + user_id + ) + .fetch_one(&pool) + .await + .expect("Failed to count") + .unwrap_or(0); + + for i in 0..5 { + let notification = NewNotification::email( + user_id, + NotificationType::PasswordReset, + format!("test{}@example.com", i), + "Test".to_string(), + "Body".to_string(), + ); + enqueue_notification(&pool, notification) + .await + .expect("Failed to enqueue"); + } + + let final_count: i64 = sqlx::query_scalar!( + "SELECT COUNT(*) FROM notification_queue WHERE status = 'pending' AND user_id = $1", + user_id + ) + .fetch_one(&pool) + .await + .expect("Failed to count") + .unwrap_or(0); + + assert_eq!(final_count - initial_count, 5); +}