use crate::domain::error::{AppError, AppResult}; use crate::domain::models::{Merchant, OrderRecord, ProductLink}; use crate::infrastructure::repositories::{MerchantRepository, OrderRepository, ProductRepository}; use crate::interfaces::http::api::RealtimeEvent; use async_trait::async_trait; use serde::Serialize; use std::sync::Arc; use tokio::sync::broadcast; use uuid::Uuid; use crate::infrastructure::db::DbPool; #[derive(Serialize)] pub struct StorefrontProfile { pub merchant_id: String, pub brand_name: String, pub business_address: Option, pub base_pincode: Option, pub upi_id: Option, pub announcement_banner: Option, pub avg_rating: f64, pub review_count: i64, } #[derive(Serialize)] pub struct CatalogItem { pub product: ProductLink, pub avg_rating: f64, pub review_count: i64, } #[async_trait] #[allow(clippy::too_many_arguments)] pub trait MerchantService: Send + Sync { async fn get_profile(&self, merchant_id: &str) -> AppResult>; async fn get_by_slug(&self, slug: &str) -> AppResult>; async fn update_profile( &self, merchant_id: &str, brand_name: Option, social_url: Option, upi_id: Option, business_address: Option, delivery_rate_per_km: Option, delivery_base_fee: Option, logistics_config: Option, base_pincode: Option, auto_settle_threshold: Option, announcement_banner: Option, ) -> AppResult<()>; async fn get_products(&self, merchant_id: &str) -> AppResult>; async fn get_orders(&self, merchant_id: &str) -> AppResult>; async fn get_order(&self, merchant_id: &str, transaction_id: &str) -> AppResult; async fn get_customers( &self, merchant_id: &str, ) -> AppResult>; async fn get_products_by_slug(&self, slug: &str) -> AppResult>; async fn get_merchant_upi(&self, merchant_id: &str) -> AppResult>; async fn approve_settlement( &self, merchant_id: &str, transaction_id: &str, return_weight: Option, request_id: Option, utr_number: Option, ) -> AppResult<()>; async fn mark_order_shipped( &self, merchant_id: &str, transaction_id: &str, shipping_method: Option, estimated_delivery_at: Option, ) -> AppResult<()>; async fn bulk_mark_orders_shipped( &self, merchant_id: &str, transaction_ids: Vec, ) -> AppResult<()>; async fn delete_link(&self, link_id: &str, merchant_id: &str) -> AppResult<()>; async fn create_link( &self, merchant_id: &str, product_name: &str, price_inr: f64, image_data: Option, expected_weight: f64, inventory_count: i32, is_unlimited: bool, category: Option, ) -> AppResult; async fn update_inventory( &self, merchant_id: &str, link_id: &str, count: i32, is_unlimited: bool, ) -> AppResult<()>; async fn upgrade_plan(&self, merchant_id: &str, plan: &str) -> AppResult<()>; async fn get_analytics( &self, merchant_id: &str, ) -> AppResult; async fn get_storefront_profile(&self, slug: &str) -> AppResult>; async fn get_catalog(&self, slug: &str) -> AppResult>; async fn reset_account(&self, merchant_id: &str) -> AppResult<()>; async fn submit_feedback( &self, merchant_id: &str, category: &str, message: &str, ) -> AppResult<()>; async fn calculate_credit_worthiness(&self, merchant_id: &str) -> AppResult; async fn get_dispute_evidence( &self, merchant_id: &str, transaction_id: &str, ) -> AppResult>; async fn upload_dispute_evidence( &self, merchant_id: &str, transaction_id: &str, evidence_url: String, ) -> AppResult; async fn resolve_dispute( &self, merchant_id: &str, transaction_id: &str, resolution: String, ) -> AppResult<()>; async fn dispute_order( &self, merchant_id: &str, transaction_id: &str, reason: String, request_id: Option, ) -> AppResult<()>; async fn delete_order(&self, merchant_id: &str, transaction_id: &str) -> AppResult<()>; async fn get_growth_insights(&self, merchant_id: &str) -> AppResult; async fn get_accounting_summary(&self, merchant_id: &str) -> AppResult; async fn generate_gst_report( &self, merchant_id: &str, month: u32, year: i32, ) -> AppResult; async fn create_coupon( &self, merchant_id: &str, coupon: crate::domain::models::Coupon, ) -> AppResult<()>; async fn get_coupons(&self, merchant_id: &str) -> AppResult>; async fn delete_coupon(&self, merchant_id: &str, coupon_id: &uuid::Uuid) -> AppResult<()>; async fn validate_coupon( &self, merchant_id: &str, code: &str, amount: f64, ) -> AppResult; async fn get_product_reviews( &self, product_id: &str, ) -> AppResult>; async fn get_order_invoice(&self, merchant_id: &str, transaction_id: &str) -> AppResult; async fn get_financial_ledger(&self, merchant_id: &str) -> AppResult>; } #[derive(Serialize)] pub struct LedgerRecord { pub transaction_id: String, pub created_at: Option, pub status: String, pub gross_amount: f64, pub platform_fee: f64, pub delivery_fee: f64, pub tax_amount: f64, pub net_settlement: f64, pub settled_at: Option, pub utr_number: Option, } pub struct RtixMerchantService { merchant_repo: Arc, product_repo: Arc, order_repo: Arc, coupon_repo: Arc, pool: DbPool, tx: broadcast::Sender, } impl RtixMerchantService { pub fn new( merchant_repo: Arc, product_repo: Arc, order_repo: Arc, coupon_repo: Arc, pool: DbPool, tx: broadcast::Sender, ) -> Self { Self { merchant_repo, product_repo, order_repo, coupon_repo, pool, tx, } } } #[async_trait] impl MerchantService for RtixMerchantService { async fn get_profile(&self, merchant_id: &str) -> AppResult> { self.merchant_repo.find_by_id(merchant_id).await } async fn get_by_slug(&self, slug: &str) -> AppResult> { self.merchant_repo.find_by_slug(slug).await } async fn update_profile( &self, merchant_id: &str, brand_name: Option, social_url: Option, upi_id: Option, business_address: Option, delivery_rate_per_km: Option, delivery_base_fee: Option, logistics_config: Option, base_pincode: Option, auto_settle_threshold: Option, announcement_banner: Option, ) -> AppResult<()> { let mut changes = Vec::new(); if brand_name.is_some() { changes.push("brand_name"); } if upi_id.is_some() { changes.push("upi_id"); } if social_url.is_some() { changes.push("social_url"); } if business_address.is_some() { changes.push("business_address"); } if base_pincode.is_some() { changes.push("base_pincode"); } if delivery_rate_per_km.is_some() { changes.push("delivery_rate_per_km"); } if delivery_base_fee.is_some() { changes.push("delivery_base_fee"); } if auto_settle_threshold.is_some() { changes.push("auto_settle_threshold"); } if announcement_banner.is_some() { changes.push("announcement_banner"); } self.merchant_repo .update_profile( merchant_id, brand_name, social_url, upi_id, business_address, delivery_rate_per_km, delivery_base_fee, logistics_config, base_pincode, auto_settle_threshold, announcement_banner, ) .await?; if let Ok(mut conn) = self.pool.acquire().await { let details = if changes.is_empty() { "Merchant profile settings updated".to_string() } else { format!("Merchant profile settings updated: {}", changes.join(", ")) }; crate::domain::audit::log_risk_event( &mut conn, None, merchant_id, "PROFILE_UPDATED", "LOW", Some(&details), None, None, None, Some(&self.tx), ) .await; } Ok(()) } async fn get_products(&self, merchant_id: &str) -> AppResult> { let profile = self.merchant_repo.find_by_id(merchant_id).await?; let is_admin = profile .map(|m| m.role == "DEVELOPER" || m.role == "ADMIN") .unwrap_or(false); let products = if is_admin { sqlx::query_as::<_, ProductLink>( "SELECT link_id, merchant_id, product_name, price_inr, image_data, expected_weight, link_views, inventory_count, is_unlimited, category, is_featured, sale_price_inr, sale_ends_at, created_at FROM product_links ORDER BY created_at DESC" ) .fetch_all(&self.pool) .await .map_err(crate::domain::error::AppError::Database)? } else { self.product_repo.all_for_merchant(merchant_id).await? }; let hydration_futures = products.into_iter().map(|mut p| async move { p.image_data = crate::core::utils::hydrate_file_to_base64(p.image_data).await; p }); let hydrated = futures_util::future::join_all(hydration_futures).await; Ok(hydrated) } async fn get_products_by_slug(&self, slug: &str) -> AppResult> { self.product_repo.find_by_slug(slug).await } async fn get_merchant_upi(&self, merchant_id: &str) -> AppResult> { let profile = self.merchant_repo.find_by_id(merchant_id).await?; Ok(profile.and_then(|p| p.upi_id)) } async fn delete_link(&self, link_id: &str, merchant_id: &str) -> AppResult<()> { let product_info = self.product_repo.find_by_id(link_id).await.ok().flatten(); let details = product_info .map(|p| format!("Product link deleted: {} (ID: {})", p.product_name, link_id)) .unwrap_or_else(|| format!("Product link deleted: {}", link_id)); self.product_repo.delete(link_id, merchant_id).await?; if let Ok(mut conn) = self.pool.acquire().await { crate::domain::audit::log_risk_event( &mut conn, None, merchant_id, "PRODUCT_DELETED", "LOW", Some(&details), None, None, None, Some(&self.tx), ) .await; } Ok(()) } async fn create_link( &self, merchant_id: &str, product_name: &str, price_inr: f64, image_data: Option, expected_weight: f64, inventory_count: i32, is_unlimited: bool, category: Option, ) -> AppResult { let merchant = self .merchant_repo .find_by_id(merchant_id) .await? .ok_or_else(|| AppError::NotFound("Merchant not found".to_string()))?; if merchant.plan == "FREE" { let active_products = self.product_repo.all_for_merchant(merchant_id).await?; if active_products.len() >= 10 { return Err(AppError::Forbidden( "Product limit reached (10). Upgrade to PRO to create unlimited product links." .to_string(), )); } } let link_id = Uuid::new_v4().to_string(); let product = ProductLink { link_id: link_id.clone(), merchant_id: merchant_id.to_string(), product_name: product_name.to_string(), price_inr, image_data, expected_weight, link_views: 0, inventory_count, is_unlimited, category, is_featured: false, sale_price_inr: None, sale_ends_at: None, created_at: Some(chrono::Utc::now().naive_utc()), }; self.product_repo.create(&product).await?; if let Ok(mut conn) = self.pool.acquire().await { crate::domain::audit::log_risk_event( &mut conn, None, merchant_id, "PRODUCT_CREATED", "LOW", Some(&format!( "Product link created: {} (Price: ₹{})", product_name, price_inr )), None, None, None, Some(&self.tx), ) .await; } Ok(product) } async fn update_inventory( &self, merchant_id: &str, link_id: &str, count: i32, is_unlimited: bool, ) -> AppResult<()> { self.product_repo .update_inventory(link_id, merchant_id, count, is_unlimited) .await?; if let Ok(mut conn) = self.pool.acquire().await { let details = if is_unlimited { format!( "Inventory updated to Unlimited for product (ID: {})", link_id ) } else { format!( "Inventory count updated to {} for product (ID: {})", count, link_id ) }; crate::domain::audit::log_risk_event( &mut conn, None, merchant_id, "PRODUCT_INVENTORY_UPDATED", "LOW", Some(&details), None, None, None, Some(&self.tx), ) .await; } Ok(()) } async fn get_orders(&self, merchant_id: &str) -> AppResult> { let profile = self.merchant_repo.find_by_id(merchant_id).await?; let is_admin = profile .map(|m| m.role == "DEVELOPER" || m.role == "ADMIN") .unwrap_or(false); if is_admin { let orders = sqlx::query_as::<_, OrderRecord>("SELECT * FROM orders ORDER BY created_at DESC") .fetch_all(&self.pool) .await .map_err(crate::domain::error::AppError::Database)?; let decrypted: Vec = orders .into_iter() .map(|mut o| { o.decrypt_pii(); if o.vpa.as_deref() == Some("") { o.vpa = None; } o }) .collect(); Ok(decrypted) } else { self.order_repo.all_for_merchant(merchant_id).await } } async fn get_order(&self, merchant_id: &str, transaction_id: &str) -> AppResult { let order = self.order_repo.find_by_id(transaction_id).await?; let order = order.ok_or_else(|| AppError::NotFound("Order not found".into()))?; let profile = self.merchant_repo.find_by_id(merchant_id).await?; let is_admin = profile .map(|m| m.role == "DEVELOPER" || m.role == "ADMIN") .unwrap_or(false); if !is_admin && order.merchant_id != merchant_id { return Err(AppError::Forbidden("Access denied".into())); } Ok(order) } async fn get_customers( &self, merchant_id: &str, ) -> AppResult> { let profile = self.merchant_repo.find_by_id(merchant_id).await?; let is_admin = profile .map(|m| m.role == "DEVELOPER" || m.role == "ADMIN") .unwrap_or(false); let orders = if is_admin { let orders = sqlx::query_as::<_, OrderRecord>("SELECT * FROM orders") .fetch_all(&self.pool) .await .map_err(crate::domain::error::AppError::Database)?; orders .into_iter() .map(|mut o| { o.decrypt_pii(); o }) .collect::>() } else { self.order_repo.all_for_merchant(merchant_id).await? }; // Group and aggregate in memory in Rust by plaintext phone use std::collections::HashMap; let mut customer_map: HashMap = HashMap::new(); for order in orders { let phone = order.buyer_phone.clone(); let price = order.price_inr; let order_date = order.created_at; if let Some(record) = customer_map.get_mut(&phone) { record.order_count += 1; record.total_spent += price; if let Some(date) = order_date { if record.last_order_date.is_none() || Some(date) > record.last_order_date { record.last_order_date = Some(date); } } } else { customer_map.insert( phone.clone(), crate::domain::models::CustomerRecord { buyer_phone: phone, order_count: 1, total_spent: price, last_order_date: order_date, }, ); } } let mut customers: Vec = customer_map.into_values().collect(); // Sort by last order date descending customers.sort_by(|a, b| b.last_order_date.cmp(&a.last_order_date)); Ok(customers) } async fn approve_settlement( &self, merchant_id: &str, transaction_id: &str, return_weight: Option, request_id: Option, utr_number: Option, ) -> AppResult<()> { let mut order = self.get_order(merchant_id, transaction_id).await?; // 1. Idempotency & State Guarding if order.status == crate::domain::constants::ORDER_STATUS_SETTLED { tracing::warn!( "Settlement IDEMPOTENCY: Order {} already settled. Skipping.", transaction_id ); return Ok(()); } let allowed_statuses = [ crate::domain::constants::ORDER_STATUS_DELIVERED_PENDING_APPROVAL, crate::domain::constants::ORDER_STATUS_DISPUTED_HELD, ]; if !allowed_statuses.contains(&order.status.as_str()) { return Err(AppError::BadRequest( format!("Invalid State Transition: Cannot settle order in '{}' status. Manual review or delivery confirmation required.", order.status), )); } // 2. Weight Correction & UTR Capture if let Some(w) = return_weight { order.return_weight = w; // Volumetric Exception Flow: Verify weight integrity (with 2% tolerance threshold) if order.outbound_weight > 0.0 && !crate::core::utils::verify_volumetric_integrity(order.outbound_weight, w, 2.0) { // Escalate to DISPUTED status and lock payout let dispute_status = crate::domain::constants::ORDER_STATUS_DISPUTED.to_string(); let mut tx = self.pool.begin().await.map_err(AppError::Database)?; sqlx::query("UPDATE orders SET status = $1, return_weight = $2 WHERE transaction_id = $3") .bind(&dispute_status) .bind(order.return_weight) .bind(transaction_id) .execute(&mut *tx) .await?; if let Some(ref rid) = request_id { crate::domain::audit::log_risk_event( &mut tx, Some(transaction_id), merchant_id, "VOLUMETRIC_MISMATCH", "HIGH", Some(&format!( "Volumetric exception: Outbound weight ({}g) and return weight ({}g) mismatch. Payout locked, dispute opened.", order.outbound_weight, w )), Some((order.outbound_weight - w).abs()), Some(rid), order.device_fingerprint.as_deref(), Some(&self.tx), ) .await; } tx.commit().await.map_err(AppError::Database)?; let _ = self.tx.send(RealtimeEvent::OrderStatusChanged { transaction_id: transaction_id.to_string(), merchant_id: merchant_id.to_string(), new_status: dispute_status.clone(), }); return Err(AppError::BadRequest(format!( "Volumetric Exception: Outbound weight ({}g) and return weight ({}g) mismatch. Payout locked, order status moved to DISPUTED.", order.outbound_weight, w ))); } } if let Some(ref utr) = utr_number { order.utr_number = Some(utr.clone()); } let original_status = order.status.clone(); order.status = crate::domain::constants::ORDER_STATUS_SETTLED.to_string(); order.settled_at = Some(chrono::Utc::now().naive_utc()); let mut tx = self.pool.begin().await.map_err(AppError::Database)?; // 3. Atomic Finality sqlx::query("UPDATE orders SET status = $1, settled_at = $2, return_weight = $3, utr_number = $4 WHERE transaction_id = $5 AND status = $6") .bind(&order.status) .bind(order.settled_at) .bind(order.return_weight) .bind(&order.utr_number) .bind(transaction_id) .bind(&original_status) .execute(&mut *tx) .await?; // 4. Institutional Ledger Entry let (gross, platform, delivery, tax, net) = order.calculate_net_settlement(); sqlx::query( r#" INSERT INTO settlements ( transaction_id, merchant_id, gross_amount_inr, platform_fee_inr, delivery_fee_inr, tax_amount_inr, net_payout_inr, utr_number ) VALUES ($1, $2, $3, $4, $5, $6, $7, $8) "#, ) .bind(transaction_id) .bind(merchant_id) .bind(gross) .bind(platform) .bind(delivery) .bind(tax) .bind(net) .bind(&order.utr_number) .execute(&mut *tx) .await?; // 5. Institutional Audit Log if let Some(rid) = request_id { crate::domain::audit::log_risk_event( &mut tx, Some(transaction_id), merchant_id, "SETTLEMENT_FINALIZED", "LOW", Some("Settlement finalized manually by merchant. Net payout calculation locked."), None, Some(&rid), order.device_fingerprint.as_deref(), Some(&self.tx), ) .await; } tx.commit().await.map_err(AppError::Database)?; // Broadcast for real-time frontend updates let _ = self.tx.send(RealtimeEvent::OrderStatusChanged { transaction_id: transaction_id.to_string(), merchant_id: merchant_id.to_string(), new_status: order.status.clone(), }); Ok(()) } async fn mark_order_shipped( &self, merchant_id: &str, transaction_id: &str, shipping_method: Option, estimated_delivery_at: Option, ) -> AppResult<()> { let order = self.get_order(merchant_id, transaction_id).await?; if order.status != crate::domain::constants::ORDER_STATUS_PAID_PENDING_DELIVERY { return Err(AppError::BadRequest( "Order is not in a state that can be shipped".to_string(), )); } sqlx::query("UPDATE orders SET status = $1, shipped_at = CURRENT_TIMESTAMP, shipping_method = $2, estimated_delivery_at = $3 WHERE transaction_id = $4") .bind(crate::domain::constants::ORDER_STATUS_DELIVERED_PENDING_APPROVAL) .bind(shipping_method) .bind(estimated_delivery_at) .bind(transaction_id) .execute(&self.pool) .await?; Ok(()) } async fn bulk_mark_orders_shipped( &self, merchant_id: &str, transaction_ids: Vec, ) -> AppResult<()> { self.order_repo .bulk_mark_shipped(merchant_id, &transaction_ids) .await } async fn delete_order(&self, merchant_id: &str, transaction_id: &str) -> AppResult<()> { let order = self.get_order(merchant_id, transaction_id).await?; self.order_repo.delete(&order.transaction_id).await } async fn dispute_order( &self, merchant_id: &str, transaction_id: &str, reason: String, request_id: Option, ) -> AppResult<()> { let order = self.get_order(merchant_id, transaction_id).await?; let mut tx = self.pool.begin().await.map_err(AppError::Database)?; sqlx::query("UPDATE orders SET status = $1 WHERE transaction_id = $2") .bind(crate::domain::constants::ORDER_STATUS_DISPUTED) .bind(&order.transaction_id) .execute(&mut *tx) .await?; if let Some(rid) = request_id { crate::domain::audit::log_risk_event( &mut tx, Some(transaction_id), merchant_id, "ORDER_DISPUTED", "MEDIUM", Some(&format!("Merchant raised dispute: {}", reason)), None, Some(&rid), order.device_fingerprint.as_deref(), Some(&self.tx), ) .await; } tx.commit().await.map_err(AppError::Database)?; Ok(()) } async fn get_storefront_profile(&self, slug: &str) -> AppResult> { let merchant = self.merchant_repo.find_by_slug(slug).await?; if let Some(m) = merchant { if m.is_frozen { return Err(AppError::Forbidden( "Merchant account is frozen due to unpaid outstanding invoices.".to_string(), )); } let stats = sqlx::query("SELECT AVG(rating)::FLOAT8 as avg_rating, COUNT(*)::BIGINT as review_count FROM product_feedback f JOIN orders o ON f.transaction_id = o.transaction_id WHERE o.merchant_id = $1 AND f.is_public = TRUE") .bind(&m.merchant_id) .fetch_one(&self.pool) .await?; use sqlx::Row; Ok(Some(StorefrontProfile { merchant_id: m.merchant_id, brand_name: m.brand_name, business_address: m.business_address, base_pincode: Some(m.base_pincode), upi_id: m.upi_id, announcement_banner: m.announcement_banner, avg_rating: stats.get::, _>("avg_rating").unwrap_or(0.0), review_count: stats.get::, _>("review_count").unwrap_or(0), })) } else { Ok(None) } } async fn get_catalog(&self, slug: &str) -> AppResult> { let products = self.product_repo.find_by_slug(slug).await?; if products.is_empty() { return Ok(vec![]); } let link_ids: Vec = products .iter() .map(|p| Uuid::parse_str(&p.link_id).unwrap_or_default()) .collect(); let stats_map = sqlx::query( "SELECT product_id, AVG(rating)::FLOAT8 as avg_rating, COUNT(*)::BIGINT as review_count \ FROM product_feedback \ WHERE product_id = ANY($1) AND is_public = TRUE \ GROUP BY product_id" ) .bind(&link_ids) .fetch_all(&self.pool) .await? .into_iter() .map(|r| { use sqlx::Row; ( r.get::("product_id"), (r.get::, _>("avg_rating").unwrap_or(0.0), r.get::, _>("review_count").unwrap_or(0)) ) }) .collect::>(); let catalog_futures = products.into_iter().map(|mut p| { let stats_map_ref = &stats_map; async move { p.image_data = crate::core::utils::hydrate_file_to_base64(p.image_data).await; let pid = Uuid::parse_str(&p.link_id).unwrap_or_default(); let (avg, count) = stats_map_ref.get(&pid).cloned().unwrap_or((0.0, 0)); CatalogItem { product: p, avg_rating: avg, review_count: count, } } }); let catalog = futures_util::future::join_all(catalog_futures).await; Ok(catalog) } async fn get_analytics( &self, merchant_id: &str, ) -> AppResult { use futures_util::TryFutureExt; use sqlx::Row; let profile = self.merchant_repo.find_by_id(merchant_id).await?; let is_admin = profile .as_ref() .map(|m| m.role == "DEVELOPER" || m.role == "ADMIN") .unwrap_or(false); let (merchant_opt, summary, total_views, risk_data, daily_rows, regional_rows) = if is_admin { tokio::try_join!( self.merchant_repo.find_by_id(merchant_id), sqlx::query( r#" SELECT COALESCE(SUM(CASE WHEN status = $1 THEN price_inr ELSE 0 END), 0)::FLOAT8 as total_revenue, COUNT(*)::BIGINT as total_orders, COALESCE(SUM(platform_fee), 0)::FLOAT8 as platform_fee_total, COALESCE(AVG(distance_km), 0)::FLOAT8 as avg_distance, COUNT(*) FILTER (WHERE risk_score > 60)::BIGINT as risk_mitigation_count FROM orders WHERE status NOT IN ('PENDING_UPI_SETTLEMENT', 'PAYMENT_FAILED') "# ) .bind(crate::domain::constants::ORDER_STATUS_SETTLED) .fetch_one(&self.pool) .map_err(crate::domain::error::AppError::Database), sqlx::query_scalar( "SELECT COALESCE(SUM(link_views), 0)::BIGINT FROM product_links", ) .fetch_one(&self.pool) .map_err(crate::domain::error::AppError::Database), sqlx::query( r#" SELECT COUNT(*) FILTER (WHERE COALESCE(risk_score, 0) <= 25)::BIGINT as low, COUNT(*) FILTER (WHERE COALESCE(risk_score, 0) > 25 AND COALESCE(risk_score, 0) <= 50)::BIGINT as medium, COUNT(*) FILTER (WHERE COALESCE(risk_score, 0) > 50 AND COALESCE(risk_score, 0) <= 75)::BIGINT as high, COUNT(*) FILTER (WHERE COALESCE(risk_score, 0) > 75)::BIGINT as critical FROM orders WHERE status NOT IN ('PENDING_UPI_SETTLEMENT', 'PAYMENT_FAILED') "# ) .fetch_one(&self.pool) .map_err(crate::domain::error::AppError::Database), sqlx::query( r#" SELECT TO_CHAR(created_at, 'YYYY-MM-DD') as day, COALESCE(SUM(price_inr), 0)::FLOAT8 as revenue, COUNT(*)::BIGINT as order_count, COUNT(*) FILTER (WHERE COALESCE(risk_score, 0) > 60)::BIGINT as high_risk_count, COALESCE(SUM(price_inr * (1.0 - (COALESCE(risk_score, 0) / 100.0))), 0)::FLOAT8 as risk_weighted_volume FROM orders WHERE status NOT IN ('PENDING_UPI_SETTLEMENT', 'PAYMENT_FAILED') AND created_at > CURRENT_DATE - INTERVAL '30 days' GROUP BY day ORDER BY day ASC "# ) .fetch_all(&self.pool) .map_err(crate::domain::error::AppError::Database), sqlx::query( r#" SELECT SUBSTRING(shipping_pincode, 1, 3) as prefix, COUNT(*)::BIGINT as order_count, COALESCE(SUM(price_inr), 0)::FLOAT8 as total_revenue, COALESCE(AVG(distance_km), 0)::FLOAT8 as avg_distance FROM orders WHERE status NOT IN ('PENDING_UPI_SETTLEMENT', 'PAYMENT_FAILED') AND shipping_pincode IS NOT NULL GROUP BY prefix ORDER BY order_count DESC LIMIT 10 "#, ) .fetch_all(&self.pool) .map_err(crate::domain::error::AppError::Database), )? } else { tokio::try_join!( self.merchant_repo.find_by_id(merchant_id), sqlx::query( r#" SELECT COALESCE(SUM(CASE WHEN status = $1 THEN price_inr ELSE 0 END), 0)::FLOAT8 as total_revenue, COUNT(*)::BIGINT as total_orders, COALESCE(SUM(platform_fee), 0)::FLOAT8 as platform_fee_total, COALESCE(AVG(distance_km), 0)::FLOAT8 as avg_distance, COUNT(*) FILTER (WHERE risk_score > 60)::BIGINT as risk_mitigation_count FROM orders WHERE merchant_id = $2 AND status NOT IN ('PENDING_UPI_SETTLEMENT', 'PAYMENT_FAILED') "# ) .bind(crate::domain::constants::ORDER_STATUS_SETTLED) .bind(merchant_id) .fetch_one(&self.pool) .map_err(crate::domain::error::AppError::Database), sqlx::query_scalar( "SELECT COALESCE(SUM(link_views), 0)::BIGINT FROM product_links WHERE merchant_id = $1", ) .bind(merchant_id) .fetch_one(&self.pool) .map_err(crate::domain::error::AppError::Database), sqlx::query( r#" SELECT COUNT(*) FILTER (WHERE COALESCE(risk_score, 0) <= 25)::BIGINT as low, COUNT(*) FILTER (WHERE COALESCE(risk_score, 0) > 25 AND COALESCE(risk_score, 0) <= 50)::BIGINT as medium, COUNT(*) FILTER (WHERE COALESCE(risk_score, 0) > 50 AND COALESCE(risk_score, 0) <= 75)::BIGINT as high, COUNT(*) FILTER (WHERE COALESCE(risk_score, 0) > 75)::BIGINT as critical FROM orders WHERE merchant_id = $1 AND status NOT IN ('PENDING_UPI_SETTLEMENT', 'PAYMENT_FAILED') "# ) .bind(merchant_id) .fetch_one(&self.pool) .map_err(crate::domain::error::AppError::Database), sqlx::query( r#" SELECT TO_CHAR(created_at, 'YYYY-MM-DD') as day, COALESCE(SUM(price_inr), 0)::FLOAT8 as revenue, COUNT(*)::BIGINT as order_count, COUNT(*) FILTER (WHERE COALESCE(risk_score, 0) > 60)::BIGINT as high_risk_count, COALESCE(SUM(price_inr * (1.0 - (COALESCE(risk_score, 0) / 100.0))), 0)::FLOAT8 as risk_weighted_volume FROM orders WHERE merchant_id = $1 AND status NOT IN ('PENDING_UPI_SETTLEMENT', 'PAYMENT_FAILED') AND created_at > CURRENT_DATE - INTERVAL '30 days' GROUP BY day ORDER BY day ASC "# ) .bind(merchant_id) .fetch_all(&self.pool) .map_err(crate::domain::error::AppError::Database), sqlx::query( r#" SELECT SUBSTRING(shipping_pincode, 1, 3) as prefix, COUNT(*)::BIGINT as order_count, COALESCE(SUM(price_inr), 0)::FLOAT8 as total_revenue, COALESCE(AVG(distance_km), 0)::FLOAT8 as avg_distance FROM orders WHERE merchant_id = $1 AND status NOT IN ('PENDING_UPI_SETTLEMENT', 'PAYMENT_FAILED') AND shipping_pincode IS NOT NULL GROUP BY prefix ORDER BY order_count DESC LIMIT 10 "#, ) .bind(merchant_id) .fetch_all(&self.pool) .map_err(crate::domain::error::AppError::Database), )? }; let merchant = merchant_opt .ok_or_else(|| crate::domain::error::AppError::NotFound("Merchant not found".into()))?; let total_revenue: f64 = summary.get("total_revenue"); let total_orders: i64 = summary.get("total_orders"); let platform_fee_total: f64 = summary.get("platform_fee_total"); let avg_distance: f64 = summary.get("avg_distance"); let risk_mitigation_count: i64 = summary.get("risk_mitigation_count"); let conversion_rate = if total_views > 0 { (total_orders as f64 / total_views as f64) * 100.0 } else { 0.0 }; let risk_distribution = crate::domain::models::analytics::RiskDistribution { low: risk_data.get::("low"), medium: risk_data.get::("medium"), high: risk_data.get::("high"), critical: risk_data.get::("critical"), }; let daily_metrics = daily_rows .into_iter() .map(|r| crate::domain::models::analytics::DailyMetric { day: r.get("day"), revenue: r.get::("revenue"), order_count: r.get::("order_count"), high_risk_count: r.get::("high_risk_count"), risk_weighted_volume: r.get::("risk_weighted_volume"), }) .collect(); let regional_metrics = regional_rows .into_iter() .map(|r| crate::domain::models::analytics::RegionalMetric { pincode_prefix: r.get::, _>("prefix").unwrap_or_default(), order_count: r.get("order_count"), total_revenue: r.get("total_revenue"), avg_distance: r.get("avg_distance"), }) .collect(); Ok(crate::domain::models::analytics::AnalyticsResponse { total_revenue, total_orders, total_views, conversion_rate, daily_metrics, regional_metrics: if merchant.plan == "PRO" { regional_metrics } else { vec![] }, risk_mitigation_count, risk_distribution, platform_fee_total, abandoned_cart_rate: 0.0, // Future: Track checkout drops average_order_value: if total_orders > 0 { total_revenue / total_orders as f64 } else { 0.0 }, avg_distance, plan: merchant.plan, }) } async fn reset_account(&self, merchant_id: &str) -> AppResult<()> { self.merchant_repo.reset_account(merchant_id).await?; if let Ok(mut conn) = self.pool.acquire().await { crate::domain::audit::log_risk_event( &mut conn, None, merchant_id, "ACCOUNT_RESET", "LOW", Some("Merchant account data has been reset to defaults."), None, None, None, Some(&self.tx), ) .await; } Ok(()) } async fn submit_feedback( &self, merchant_id: &str, category: &str, message: &str, ) -> AppResult<()> { sqlx::query("INSERT INTO feedback (merchant_id, category, message) VALUES ($1, $2, $3)") .bind(merchant_id) .bind(category) .bind(message) .execute(&self.pool) .await?; if let Ok(mut conn) = self.pool.acquire().await { crate::domain::audit::log_risk_event( &mut conn, None, merchant_id, "FEEDBACK_SUBMITTED", "LOW", Some(&format!( "Merchant feedback submitted for category: {}", category )), None, None, None, Some(&self.tx), ) .await; } Ok(()) } async fn upgrade_plan(&self, merchant_id: &str, plan: &str) -> AppResult<()> { sqlx::query("UPDATE merchants SET plan = $1 WHERE merchant_id = $2") .bind(plan) .bind(merchant_id) .execute(&self.pool) .await?; tracing::info!("Merchant {} upgraded to {} plan", merchant_id, plan); if let Ok(mut conn) = self.pool.acquire().await { crate::domain::audit::log_risk_event( &mut conn, None, merchant_id, "PLAN_UPGRADED", "LOW", Some(&format!("Subscription plan upgraded to {}", plan)), None, None, None, Some(&self.tx), ) .await; } Ok(()) } async fn calculate_credit_worthiness(&self, merchant_id: &str) -> AppResult { let orders = sqlx::query( "SELECT COUNT(*) as total, \ COUNT(*) FILTER (WHERE status = $1) as settled, \ COUNT(*) FILTER (WHERE status = $2) as disputed, \ SUM(price_inr) FILTER (WHERE status = $1) as total_revenue \ FROM orders WHERE merchant_id = $3 AND status NOT IN ('PENDING_UPI_SETTLEMENT', 'PAYMENT_FAILED')", ) .bind(crate::domain::constants::ORDER_STATUS_SETTLED) .bind(crate::domain::constants::ORDER_STATUS_DISPUTED) .bind(merchant_id) .fetch_one(&self.pool) .await?; use sqlx::Row; let total: i64 = orders.get("total"); let settled: i64 = orders.get("settled"); let disputed: i64 = orders.get("disputed"); let revenue: f64 = orders.get::, _>("total_revenue").unwrap_or(0.0); let settlement_rate = if total > 0 { settled as f64 / total as f64 } else { 0.0 }; let dispute_rate = if total > 0 { disputed as f64 / total as f64 } else { 0.0 }; // Base Score (Starting at 300) let mut score = 300.0; // Volume Bonus score += (total as f64 * 2.0).min(200.0); // Settlement Reliability Bonus score += settlement_rate * 300.0; // Dispute Penalty score -= dispute_rate * 500.0; // Revenue Multiplier score += (revenue / 10000.0).min(100.0); let final_score = score.clamp(300.0, 900.0); let tier = if final_score > 800.0 { "PLATINUM" } else if final_score > 650.0 { "GOLD" } else { "SILVER" }; Ok(serde_json::json!({ "score": final_score.round(), "tier": tier, "metrics": { "settlement_reliability": settlement_rate * 100.0, "dispute_incidence": dispute_rate * 100.0, "total_reconciled_volume": revenue }, "credit_limit_multiplier": if tier == "PLATINUM" { 5.0 } else if tier == "GOLD" { 2.5 } else { 1.0 } })) } async fn get_dispute_evidence( &self, _merchant_id: &str, transaction_id: &str, ) -> AppResult> { let evidences = sqlx::query_as::<_, crate::application::services::arbitration::DisputeEvidence>( "SELECT evidence_id, transaction_id, evidence_url, uploader_role, metadata FROM dispute_evidence WHERE transaction_id = $1" ) .bind(transaction_id) .fetch_all(&self.pool) .await?; Ok(evidences) } async fn upload_dispute_evidence( &self, _merchant_id: &str, transaction_id: &str, evidence_url: String, ) -> AppResult { let arbitration = crate::application::services::arbitration::ArbitrationService::new(self.pool.clone()); arbitration .upload_evidence(transaction_id, &evidence_url, "MERCHANT") .await } async fn resolve_dispute( &self, _merchant_id: &str, transaction_id: &str, resolution: String, ) -> AppResult<()> { let arbitration = crate::application::services::arbitration::ArbitrationService::new(self.pool.clone()); arbitration .resolve_dispute(transaction_id, &resolution) .await } async fn get_growth_insights(&self, merchant_id: &str) -> AppResult { let top_products = sqlx::query( "SELECT p.product_name, COUNT(o.transaction_id) as order_count, SUM(o.price_inr) as revenue \ FROM product_links p \ LEFT JOIN orders o ON p.link_id = o.link_id \ WHERE p.merchant_id = $1 AND (o.status = $2 OR o.status IS NULL) \ GROUP BY p.product_name \ ORDER BY revenue DESC NULLS LAST \ LIMIT 5" ) .bind(merchant_id) .bind(crate::domain::constants::ORDER_STATUS_SETTLED) .fetch_all(&self.pool) .await?; let coupon_usage = sqlx::query( "SELECT coupon_code, COUNT(*) as usage_count, SUM(discount_amount) as total_savings \ FROM orders \ WHERE merchant_id = $1 AND status NOT IN ('PENDING_UPI_SETTLEMENT', 'PAYMENT_FAILED') AND coupon_code IS NOT NULL \ GROUP BY coupon_code \ ORDER BY usage_count DESC", ) .bind(merchant_id) .fetch_all(&self.pool) .await?; use sqlx::Row; let daily_revenue = sqlx::query( "SELECT DATE(created_at) as date, SUM(price_inr) as revenue \ FROM orders \ WHERE merchant_id = $1 AND status = $2 AND created_at > CURRENT_DATE - INTERVAL '30 days' \ GROUP BY DATE(created_at) \ ORDER BY date ASC" ) .bind(merchant_id) .bind(crate::domain::constants::ORDER_STATUS_SETTLED) .fetch_all(&self.pool) .await?; let revenues: Vec = daily_revenue .iter() .map(|r| r.get::("revenue")) .collect(); let avg_daily_rev = if !revenues.is_empty() { revenues.iter().sum::() / revenues.len() as f64 } else { 0.0 }; let forecasted_revenue = avg_daily_rev * 30.0; let volatility = if revenues.len() > 1 { let mean = avg_daily_rev; let variance = revenues.iter().map(|&v| (v - mean).powi(2)).sum::() / revenues.len() as f64; variance.sqrt() / mean.max(1.0) } else { 0.5 }; let confidence = (1.0 - volatility).clamp(0.0, 1.0) * 100.0; Ok(serde_json::json!({ "top_products": top_products.into_iter().map(|p| serde_json::json!({ "name": p.get::("product_name"), "orders": p.get::("order_count"), "revenue": p.get::, _>("revenue").unwrap_or(0.0) })).collect::>(), "coupon_performance": coupon_usage.into_iter().map(|c| serde_json::json!({ "code": c.get::("coupon_code"), "usage": c.get::("usage_count"), "savings": c.get::, _>("total_savings").unwrap_or(0.0) })).collect::>(), "forecasting": { "next_30_days_projected_revenue": forecasted_revenue, "confidence_score": confidence, "growth_velocity": if avg_daily_rev > 0.0 { "STABLE" } else { "IDLE" } }, "retention_rate": 0.0, })) } async fn get_accounting_summary(&self, _merchant_id: &str) -> AppResult { Ok(serde_json::json!({ "total_payouts": 0.0, "pending_settlements": 0.0, "tax_liability": 0.0 })) } async fn generate_gst_report( &self, merchant_id: &str, month: u32, year: i32, ) -> AppResult { let orders = sqlx::query_as::<_, crate::domain::models::OrderRecord>( "SELECT * FROM orders WHERE merchant_id = $1 AND EXTRACT(MONTH FROM created_at) = $2 AND EXTRACT(YEAR FROM created_at) = $3 AND status = $4" ) .bind(merchant_id) .bind(month as f64) .bind(year as f64) .bind(crate::domain::constants::ORDER_STATUS_SETTLED) .fetch_all(&self.pool) .await?; let mut csv = "Date,Transaction ID,Buyer,Gross Price,CGST,SGST,IGST,Total Tax,Net Settlement\n" .to_string(); for o in orders { let tax = o.cgst + o.sgst + o.igst; let net = o.price_inr - (o.platform_fee + o.delivery_fee); csv.push_str(&format!( "{},{},{},{:.2},{:.2},{:.2},{:.2},{:.2},{:.2}\n", o.created_at .map(|d| d.date().to_string()) .unwrap_or_default(), o.transaction_id, o.buyer_name, o.price_inr, o.cgst, o.sgst, o.igst, tax, net )); } Ok(csv) } async fn create_coupon( &self, merchant_id: &str, coupon: crate::domain::models::Coupon, ) -> AppResult<()> { let code = coupon.code.clone(); let val = coupon.discount_value; let dtype = coupon.discount_type.clone(); self.coupon_repo.create(&coupon).await?; if let Ok(mut conn) = self.pool.acquire().await { crate::domain::audit::log_risk_event( &mut conn, None, merchant_id, "COUPON_CREATED", "LOW", Some(&format!( "Coupon created: {} (Discount: {} {})", code, val, dtype )), None, None, None, Some(&self.tx), ) .await; } Ok(()) } async fn get_coupons( &self, merchant_id: &str, ) -> AppResult> { self.coupon_repo.all_for_merchant(merchant_id).await } async fn delete_coupon(&self, merchant_id: &str, coupon_id: &uuid::Uuid) -> AppResult<()> { self.coupon_repo.delete(merchant_id, coupon_id).await?; if let Ok(mut conn) = self.pool.acquire().await { crate::domain::audit::log_risk_event( &mut conn, None, merchant_id, "COUPON_DELETED", "LOW", Some(&format!("Coupon deleted: {}", coupon_id)), None, None, None, Some(&self.tx), ) .await; } Ok(()) } async fn validate_coupon( &self, merchant_id: &str, code: &str, amount: f64, ) -> AppResult { let coupon = self.coupon_repo.find_by_code(merchant_id, code).await?; let coupon = coupon.ok_or_else(|| AppError::NotFound("Coupon not found or inactive".into()))?; if !coupon.is_valid(amount) { return Err(AppError::BadRequest( "Coupon conditions not met or expired".into(), )); } Ok(coupon) } async fn get_product_reviews( &self, product_id: &str, ) -> AppResult> { let reviews = sqlx::query_as::<_, crate::domain::models::ProductFeedback>( "SELECT * FROM product_feedback WHERE product_id = $1 AND is_public = TRUE ORDER BY created_at DESC" ) .bind(product_id) .fetch_all(&self.pool) .await?; Ok(reviews) } async fn get_order_invoice( &self, merchant_id: &str, transaction_id: &str, ) -> AppResult { let order = self .order_repo .find_by_id(transaction_id) .await? .ok_or_else(|| AppError::NotFound("Order not found".into()))?; if order.merchant_id != merchant_id { return Err(AppError::Forbidden("Access denied to this order".into())); } let merchant = self .merchant_repo .find_by_id(&order.merchant_id) .await? .ok_or_else(|| AppError::NotFound("Merchant not found".into()))?; Ok( crate::application::services::billing::BillingService::generate_invoice_html( &order, &merchant, ), ) } async fn get_financial_ledger(&self, merchant_id: &str) -> AppResult> { let orders = self.order_repo.all_for_merchant(merchant_id).await?; let ledger = orders .into_iter() .map(|order| { let (gross, platform, delivery, tax, net) = order.calculate_net_settlement(); LedgerRecord { transaction_id: order.transaction_id, created_at: order.created_at, status: order.status, gross_amount: gross, platform_fee: platform, delivery_fee: delivery, tax_amount: tax, net_settlement: net, settled_at: order.settled_at, utr_number: order.utr_number, } }) .collect(); Ok(ledger) } }