use crate::infrastructure::db::DbPool; use crate::interfaces::http::api::RealtimeEvent; use reqwest::Client; use std::sync::Arc; use tokio::sync::broadcast; use tracing::{error, info, warn}; use uuid::Uuid; use sqlx::Row; pub struct SmsNotificationService { pool: DbPool, client: Client, frontend_url: String, } impl SmsNotificationService { pub fn new(pool: DbPool) -> Self { let frontend_url = std::env::var("FRONTEND_URL") .unwrap_or_else(|_| "https://rtix.app".to_string()); Self { pool, client: Client::builder() .timeout(std::time::Duration::from_secs(10)) .build() .unwrap(), frontend_url, } } pub async fn run(&self, mut rx: broadcast::Receiver) { info!("Sms/WhatsApp Notification Service started."); loop { match rx.recv().await { Ok(event) => { let service = Arc::new(Self { pool: self.pool.clone(), client: self.client.clone(), frontend_url: self.frontend_url.clone(), }); tokio::spawn(async move { if let Err(e) = service.process_event(event).await { error!("Error processing notification: {:?}", e); } }); } Err(broadcast::error::RecvError::Lagged(n)) => { warn!("Notification Service lagged by {} events", n); } Err(broadcast::error::RecvError::Closed) => break, } } } async fn process_event( &self, event: RealtimeEvent, ) -> Result<(), Box> { match event { RealtimeEvent::NewOrder { transaction_id, merchant_id, amount, buyer_phone, } => { let msg = format!( "Your Rtix order for ₹{:.2} has been placed. Pay or track here: {}/track/{}", amount, self.frontend_url, transaction_id ); self.send_sms(&buyer_phone, Some(&merchant_id), "SMS_ORDER_CREATED", &msg) .await?; } RealtimeEvent::OrderStatusChanged { transaction_id, merchant_id, new_status, } => { // Fetch buyer phone number from the database using dynamic query let record = sqlx::query("SELECT buyer_phone FROM orders WHERE transaction_id = $1") .bind(&transaction_id) .fetch_optional(&self.pool) .await?; if let Some(row) = record { let buyer_phone: String = row.get("buyer_phone"); let msg = format!( "Status Update: Your Rtix order {} is now {}. Track here: {}/track/{}", transaction_id, new_status, self.frontend_url, transaction_id ); self.send_sms(&buyer_phone, Some(&merchant_id), "SMS_STATUS_CHANGED", &msg) .await?; } } RealtimeEvent::NewChatMessage { transaction_id, merchant_id, message, } => { // message is a JSON object matching ChatMessage struct let sender_type = message.get("sender_type").and_then(|v| v.as_str()).unwrap_or(""); let content = message.get("content").and_then(|v| v.as_str()).unwrap_or(""); if sender_type == "MERCHANT" { // Notify buyer let record = sqlx::query("SELECT buyer_phone FROM orders WHERE transaction_id = $1") .bind(&transaction_id) .fetch_optional(&self.pool) .await?; if let Some(row) = record { let buyer_phone: String = row.get("buyer_phone"); let msg = format!( "New message from Merchant: '{}'. Reply here: {}/track/{}", content, self.frontend_url, transaction_id ); self.send_sms(&buyer_phone, Some(&merchant_id), "SMS_CHAT_MESSAGE", &msg) .await?; } } else if sender_type == "BUYER" { // Notify merchant via email/log let record = sqlx::query("SELECT email FROM merchants WHERE merchant_id = $1") .bind(&merchant_id) .fetch_optional(&self.pool) .await?; if let Some(row) = record { let email: String = row.get("email"); let msg = format!( "New message from Buyer for order {}: '{}'. Reply in dashboard: {}/dashboard", transaction_id, content, self.frontend_url ); // Log this to the notifications table as an email notification let log_id = Uuid::new_v4().to_string(); sqlx::query( r#"INSERT INTO notification_log (id, merchant_id, recipient_email, event_type, subject, status, provider_id) VALUES ($1, $2, $3, $4, $5, $6, $7)"#, ) .bind(&log_id) .bind(&merchant_id) .bind(&email) .bind("EMAIL_CHAT_MESSAGE") .bind(&msg) .bind("SENT") .bind("mock-email-provider") .execute(&self.pool) .await?; info!( recipient = %email, "Email notification logged: {}", msg ); } } } _ => {} } Ok(()) } async fn send_sms( &self, phone_number: &str, merchant_id: Option<&str>, event_type: &str, message: &str, ) -> Result<(), sqlx::Error> { let mut plan = "FREE".to_string(); if let Some(m_id) = merchant_id { if let Ok(Some(row)) = sqlx::query("SELECT plan FROM merchants WHERE merchant_id = $1") .bind(m_id) .fetch_optional(&self.pool) .await { plan = row.get("plan"); } } if plan == "FREE" { let log_id = Uuid::new_v4().to_string(); let formatted_phone = format!("sms:{}", phone_number); sqlx::query( r#"INSERT INTO notification_log (id, merchant_id, recipient_email, event_type, subject, status, provider_id, error_message) VALUES ($1, $2, $3, $4, $5, 'FAILED', $6, $7)"#, ) .bind(&log_id) .bind(merchant_id) .bind(&formatted_phone) .bind(event_type) .bind(message) .bind("mock-sms-provider") .bind("SMS notifications are disabled on FREE plan. Upgrade to STARTER or PRO plan to enable SMS.") .execute(&self.pool) .await?; info!( recipient = phone_number, "SMS skipped (FREE plan): {}", message ); return Ok(()); } let account_sid = std::env::var("TWILIO_ACCOUNT_SID").ok(); let auth_token = std::env::var("TWILIO_AUTH_TOKEN").ok(); let from_number = std::env::var("TWILIO_FROM_NUMBER").ok(); let mut provider_id = None; let mut status = "SENT"; let mut error_message = None; if let (Some(sid), Some(token), Some(from)) = (account_sid, auth_token, from_number) { let url = format!( "https://api.twilio.com/2010-04-01/Accounts/{}/Messages.json", sid ); let params = [ ("To", phone_number), ("From", &from), ("Body", message), ]; match self .client .post(&url) .basic_auth(&sid, Some(&token)) .form(¶ms) .send() .await { Ok(resp) => { let status_code = resp.status(); if status_code.is_success() { #[derive(serde::Deserialize)] struct TwilioResponse { sid: String, } if let Ok(data) = resp.json::().await { provider_id = Some(data.sid); } else { provider_id = Some("twilio-unknown-sid".to_string()); } info!( recipient = phone_number, event = event_type, "SMS sent successfully via Twilio" ); } else { status = "FAILED"; let err_text = resp.text().await.unwrap_or_else(|_| "Unknown error".to_string()); error!( recipient = phone_number, status = %status_code, "Twilio error response: {}", err_text ); error_message = Some(err_text); } } Err(e) => { status = "FAILED"; let err_msg = e.to_string(); error!("HTTP request to Twilio failed: {}", err_msg); error_message = Some(err_msg); } } } else { // Mock sending via external provider (Dry Run) info!( recipient = phone_number, event = event_type, "[DRY RUN] SMS notification: {}", message ); provider_id = Some("mock-sms-provider".to_string()); } let log_id = Uuid::new_v4().to_string(); let formatted_phone = format!("sms:{}", phone_number); sqlx::query( r#"INSERT INTO notification_log (id, merchant_id, recipient_email, event_type, subject, status, provider_id, error_message) VALUES ($1, $2, $3, $4, $5, $6, $7, $8)"#, ) .bind(&log_id) .bind(merchant_id) .bind(&formatted_phone) .bind(event_type) .bind(message) .bind(status) .bind(provider_id) .bind(error_message) .execute(&self.pool) .await?; Ok(()) } }