mirror of
https://tangled.org/tranquil.farm/tranquil-pds
synced 2026-09-04 09:16:54 +00:00
Initial notification sender service
This commit is contained in:
lewis
@@ -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
|
||||
|
||||
+15
@@ -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"
|
||||
}
|
||||
+53
@@ -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"
|
||||
}
|
||||
+14
@@ -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"
|
||||
}
|
||||
+46
@@ -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"
|
||||
}
|
||||
+53
@@ -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"
|
||||
}
|
||||
+150
@@ -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"
|
||||
}
|
||||
+1
-1
@@ -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"] }
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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';
|
||||
@@ -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 {
|
||||
|
||||
@@ -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()
|
||||
}
|
||||
|
||||
@@ -1,5 +1,6 @@
|
||||
pub mod api;
|
||||
pub mod auth;
|
||||
pub mod notifications;
|
||||
pub mod repo;
|
||||
pub mod state;
|
||||
pub mod storage;
|
||||
|
||||
+54
-3
@@ -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<bool>) {
|
||||
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();
|
||||
}
|
||||
|
||||
@@ -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,
|
||||
};
|
||||
@@ -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<Self> {
|
||||
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(())
|
||||
}
|
||||
}
|
||||
@@ -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<NotificationChannel, Arc<dyn NotificationSender>>,
|
||||
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<S: NotificationSender + 'static>(mut self, sender: S) -> Self {
|
||||
self.senders.insert(sender.channel(), Arc::new(sender));
|
||||
self
|
||||
}
|
||||
|
||||
pub async fn enqueue(&self, notification: NewNotification) -> Result<Uuid, sqlx::Error> {
|
||||
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<bool>) {
|
||||
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::<Vec<_>>(),
|
||||
"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<Vec<QueuedNotification>, 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<Uuid, sqlx::Error> {
|
||||
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<Uuid, sqlx::Error> {
|
||||
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<Uuid, sqlx::Error> {
|
||||
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<Uuid, sqlx::Error> {
|
||||
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<Uuid, sqlx::Error> {
|
||||
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<Uuid, sqlx::Error> {
|
||||
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
|
||||
}
|
||||
@@ -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<String>,
|
||||
pub body: String,
|
||||
pub metadata: Option<serde_json::Value>,
|
||||
pub attempts: i32,
|
||||
pub max_attempts: i32,
|
||||
pub last_error: Option<String>,
|
||||
pub created_at: DateTime<Utc>,
|
||||
pub updated_at: DateTime<Utc>,
|
||||
pub scheduled_for: DateTime<Utc>,
|
||||
pub processed_at: Option<DateTime<Utc>>,
|
||||
}
|
||||
|
||||
pub struct NewNotification {
|
||||
pub user_id: Uuid,
|
||||
pub channel: NotificationChannel,
|
||||
pub notification_type: NotificationType,
|
||||
pub recipient: String,
|
||||
pub subject: Option<String>,
|
||||
pub body: String,
|
||||
pub metadata: Option<serde_json::Value>,
|
||||
}
|
||||
|
||||
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,
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -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);
|
||||
}
|
||||
Reference in New Issue
Block a user