use crate::interfaces::http::api::{AppState, RealtimeEvent}; use crate::domain::models::OrderRecord; use axum::{ extract::{Path, State}, http::StatusCode, Json, Router, routing::{get, post}, }; use serde::{Deserialize, Serialize}; use chrono::NaiveDateTime; #[derive(Serialize, Deserialize, sqlx::FromRow, Clone, Debug)] pub struct ChatMessage { pub id: uuid::Uuid, pub transaction_id: String, pub sender_type: String, // "BUYER", "MERCHANT", "SYSTEM" pub content: String, pub message_type: String, // "TEXT", "REFUND_OFFER", etc. pub metadata: serde_json::Value, pub created_at: NaiveDateTime, } #[derive(Deserialize)] pub struct SendMessagePayload { pub content: String, pub message_type: String, pub metadata: Option, } pub fn router() -> Router { Router::new() .route("/:transaction_id/chat/messages", get(get_chat_messages)) .route("/:transaction_id/chat/send", post(send_chat_message)) } pub async fn get_chat_messages( State(state): State, Path(transaction_id): Path, headers: axum::http::HeaderMap, ) -> Result>, StatusCode> { // Look up order let order = OrderRecord::find_by_id(&state.pool, &transaction_id) .await .map_err(|_| StatusCode::INTERNAL_SERVER_ERROR)? .ok_or(StatusCode::NOT_FOUND)?; // Auth verification let is_authorized = if let Some(auth_header) = headers.get("Authorization").and_then(|h| h.to_str().ok()) { if let Some(token) = auth_header.strip_prefix("Bearer ") { let validation = jsonwebtoken::Validation::default(); if let Ok(token_data) = jsonwebtoken::decode::( token, &crate::core::session::decoding_key(), &validation, ) { let claims = token_data.claims; claims.sub == order.merchant_id } else { false } } else { false } } else { true }; if !is_authorized { return Err(StatusCode::FORBIDDEN); } let messages = sqlx::query_as::<_, ChatMessage>( "SELECT id, transaction_id, sender_type, content, message_type, metadata, created_at FROM chat_messages WHERE transaction_id = $1 ORDER BY created_at ASC" ) .bind(&transaction_id) .fetch_all(&state.pool) .await .map_err(|e| { tracing::error!("Failed to fetch chat messages: {:?}", e); StatusCode::INTERNAL_SERVER_ERROR })?; Ok(Json(messages)) } pub async fn send_chat_message( State(state): State, Path(transaction_id): Path, headers: axum::http::HeaderMap, Json(payload): Json, ) -> Result, StatusCode> { // Look up order let order = OrderRecord::find_by_id(&state.pool, &transaction_id) .await .map_err(|_| StatusCode::INTERNAL_SERVER_ERROR)? .ok_or(StatusCode::NOT_FOUND)?; // Auth verification and sender identification let sender_type = if let Some(auth_header) = headers.get("Authorization").and_then(|h| h.to_str().ok()) { if let Some(token) = auth_header.strip_prefix("Bearer ") { let validation = jsonwebtoken::Validation::default(); if let Ok(token_data) = jsonwebtoken::decode::( token, &crate::core::session::decoding_key(), &validation, ) { let claims = token_data.claims; if claims.sub == order.merchant_id { "MERCHANT".to_string() } else { return Err(StatusCode::FORBIDDEN); } } else { return Err(StatusCode::UNAUTHORIZED); } } else { return Err(StatusCode::BAD_REQUEST); } } else { "BUYER".to_string() }; let id = uuid::Uuid::new_v4(); let metadata = payload.metadata.unwrap_or_else(|| serde_json::json!({})); sqlx::query( "INSERT INTO chat_messages (id, transaction_id, sender_type, content, message_type, metadata) VALUES ($1, $2, $3, $4, $5, $6)" ) .bind(&id) .bind(&transaction_id) .bind(&sender_type) .bind(&payload.content) .bind(&payload.message_type) .bind(&metadata) .execute(&state.pool) .await .map_err(|e| { tracing::error!("Failed to insert chat message: {:?}", e); StatusCode::INTERNAL_SERVER_ERROR })?; let created_msg = sqlx::query_as::<_, ChatMessage>( "SELECT id, transaction_id, sender_type, content, message_type, metadata, created_at FROM chat_messages WHERE id = $1" ) .bind(&id) .fetch_one(&state.pool) .await .map_err(|_| StatusCode::INTERNAL_SERVER_ERROR)?; // Broadcast new message event let event = RealtimeEvent::NewChatMessage { transaction_id: transaction_id.clone(), merchant_id: order.merchant_id.clone(), message: serde_json::to_value(&created_msg).unwrap_or_default(), }; let _ = state.tx.send(event); // If message is refund accepted, perform order state transition if payload.message_type == "REFUND_ACCEPTED" { if let Err(e) = crate::application::services::payment::OrderStatusMachine::validate_transition(&order.status, "REFUNDED") { tracing::error!("Invalid state transition: {:?}", e); return Err(StatusCode::BAD_REQUEST); } sqlx::query("UPDATE orders SET status = 'REFUNDED' WHERE transaction_id = $1") .bind(&transaction_id) .execute(&state.pool) .await .map_err(|_| StatusCode::INTERNAL_SERVER_ERROR)?; // Set associated settlements status to 'REVERSED' to adjust ledger balance sqlx::query("UPDATE settlements SET status = 'REVERSED' WHERE transaction_id = $1") .bind(&transaction_id) .execute(&state.pool) .await .map_err(|e| { tracing::error!("Failed to reverse settlements: {:?}", e); StatusCode::INTERNAL_SERVER_ERROR })?; let _ = state.tx.send(RealtimeEvent::OrderStatusChanged { transaction_id: transaction_id.clone(), merchant_id: order.merchant_id.clone(), new_status: "REFUNDED".to_string(), }); } // AI Mediator logic let trigger_ai = payload.message_type == "AI_MEDIATOR_REQUEST" || payload.content.to_lowercase().contains("ai help"); if trigger_ai { let pool_clone = state.pool.clone(); let tx_id_clone = transaction_id.clone(); let tx_sender = state.tx.clone(); let merchant_id_clone = order.merchant_id.clone(); tokio::spawn(async move { tokio::time::sleep(std::time::Duration::from_secs(2)).await; if let Ok(Some(order)) = crate::domain::models::OrderRecord::find_by_id(&pool_clone, &tx_id_clone).await { let content = match order.status.as_str() { crate::domain::constants::ORDER_STATUS_PAID_PENDING_DELIVERY => { "Rtix Guard AI Mediator: Order is PAID_PENDING_DELIVERY. The buyer is waiting for shipment. Merchant, please update shipment status or provide details." } crate::domain::constants::ORDER_STATUS_DELIVERED_PENDING_APPROVAL => { "Rtix Guard AI Mediator: Carrier reports DELIVERED. Buyer, please confirm if the item matches your expectations, or request assistance." } crate::domain::constants::ORDER_STATUS_DISPUTED | crate::domain::constants::ORDER_STATUS_DISPUTED_HELD => { "Rtix Guard AI Mediator: Order is DISPUTED. Try offering an Interactive Card (Refund Agreement or Discount Coupon) below to resolve this quickly." } "REFUNDED" => { "Rtix Guard AI Mediator: Order is REFUNDED. Mediation complete." } _ => { "Rtix Guard AI Mediator: Active monitoring enabled. Use Interactive Card options (Refund/Coupon) to finalize dispute resolutions." } }; let ai_msg_id = uuid::Uuid::new_v4(); if let Ok(_) = sqlx::query( "INSERT INTO chat_messages (id, transaction_id, sender_type, content, message_type) VALUES ($1, $2, 'SYSTEM', $3, 'AI_MEDIATOR_RESPONSE')" ) .bind(&ai_msg_id) .bind(&tx_id_clone) .bind(content) .execute(&pool_clone) .await { if let Ok(ai_msg) = sqlx::query_as::<_, ChatMessage>( "SELECT id, transaction_id, sender_type, content, message_type, metadata, created_at FROM chat_messages WHERE id = $1" ) .bind(&ai_msg_id) .fetch_one(&pool_clone) .await { let _ = tx_sender.send(RealtimeEvent::NewChatMessage { transaction_id: tx_id_clone, merchant_id: merchant_id_clone, message: serde_json::to_value(&ai_msg).unwrap_or_default(), }); } } } }); } Ok(Json(created_msg)) }