diff --git a/services/integration_tests/tests/backtesting_service_e2e.rs b/services/integration_tests/tests/backtesting_service_e2e.rs index 427a88cb7..6d333b204 100644 --- a/services/integration_tests/tests/backtesting_service_e2e.rs +++ b/services/integration_tests/tests/backtesting_service_e2e.rs @@ -73,7 +73,7 @@ fn generate_test_token(user_id: &str, roles: Vec, permissions: Vec Result> { +async fn create_authenticated_client() -> Result) -> Result, tonic::Status> + Clone>>> { let token = generate_test_token( "test_backtester_001", vec!["analyst".to_string()], @@ -88,7 +88,17 @@ async fn create_authenticated_client() -> Result| -> Result, tonic::Status> { + let token_value = format!("Bearer {}", token); + let metadata_value = MetadataValue::try_from(token_value) + .map_err(|_| tonic::Status::internal("Failed to create metadata value"))?; + + req.metadata_mut().insert("authorization", metadata_value); + Ok(req) + }; + + let client = BacktestingServiceClient::with_interceptor(channel, interceptor); Ok(client) } diff --git a/services/integration_tests/tests/ml_training_service_e2e.rs b/services/integration_tests/tests/ml_training_service_e2e.rs index 6fb0ea2c1..0a6d2a6c8 100644 --- a/services/integration_tests/tests/ml_training_service_e2e.rs +++ b/services/integration_tests/tests/ml_training_service_e2e.rs @@ -76,7 +76,7 @@ fn generate_test_token(user_id: &str, roles: Vec, permissions: Vec Result> { +async fn create_authenticated_client() -> Result) -> Result, tonic::Status> + Clone>>> { let token = generate_test_token( "test_ml_engineer_001", vec!["ml_engineer".to_string()], @@ -92,7 +92,17 @@ async fn create_authenticated_client() -> Result| -> Result, tonic::Status> { + let token_value = format!("Bearer {}", token); + let metadata_value = MetadataValue::try_from(token_value) + .map_err(|_| tonic::Status::internal("Failed to create metadata value"))?; + + req.metadata_mut().insert("authorization", metadata_value); + Ok(req) + }; + + let client = MlTrainingServiceClient::with_interceptor(channel, interceptor); Ok(client) } diff --git a/services/integration_tests/tests/service_health_resilience_e2e.rs b/services/integration_tests/tests/service_health_resilience_e2e.rs index 51ea9bfbc..6b0f72e07 100644 --- a/services/integration_tests/tests/service_health_resilience_e2e.rs +++ b/services/integration_tests/tests/service_health_resilience_e2e.rs @@ -71,6 +71,18 @@ fn generate_test_token(user_id: &str, roles: Vec, permissions: Vec impl Fn(Request<()>) -> Result, tonic::Status> + Clone { + move |mut req: Request<()>| -> Result, tonic::Status> { + let token_value = format!("Bearer {}", token); + let metadata_value = MetadataValue::try_from(token_value) + .map_err(|_| tonic::Status::internal("Failed to create metadata value"))?; + + req.metadata_mut().insert("authorization", metadata_value); + Ok(req) + } +} + // ============================================================================ // SECTION 1: SYSTEM HEALTH MONITORING E2E TESTS (5 tests) // ============================================================================ @@ -90,7 +102,8 @@ async fn test_e2e_system_health_all_services() -> Result<()> { .connect() .await?; - let mut client = TradingServiceClient::new(channel); + let interceptor = create_auth_interceptor(token); + let mut client = TradingServiceClient::with_interceptor(channel, interceptor); let request = Request::new(GetSystemStatusRequest { service_names: vec![], // Empty = all services @@ -134,7 +147,8 @@ async fn test_e2e_system_health_specific_service() -> Result<()> { .connect() .await?; - let mut client = TradingServiceClient::new(channel); + let interceptor = create_auth_interceptor(token); + let mut client = TradingServiceClient::with_interceptor(channel, interceptor); let request = Request::new(GetSystemStatusRequest { service_names: vec!["trading_service".to_string()], @@ -171,7 +185,8 @@ async fn test_e2e_health_check_interval() -> Result<()> { .connect() .await?; - let mut client = TradingServiceClient::new(channel); + let interceptor = create_auth_interceptor(token); + let mut client = TradingServiceClient::with_interceptor(channel, interceptor); // Get initial health status let request1 = Request::new(GetSystemStatusRequest { @@ -215,7 +230,8 @@ async fn test_e2e_health_status_transitions() -> Result<()> { .connect() .await?; - let mut client = TradingServiceClient::new(channel); + let interceptor = create_auth_interceptor(token); + let mut client = TradingServiceClient::with_interceptor(channel, interceptor); // Subscribe to system status changes let request = Request::new(trading::SubscribeSystemStatusRequest { @@ -267,7 +283,8 @@ async fn test_e2e_degraded_service_detection() -> Result<()> { .connect() .await?; - let mut client = TradingServiceClient::new(channel); + let interceptor = create_auth_interceptor(token); + let mut client = TradingServiceClient::with_interceptor(channel, interceptor); let request = Request::new(GetSystemStatusRequest { service_names: vec![], @@ -316,7 +333,8 @@ async fn test_e2e_trading_service_available_backtesting_optional() -> Result<()> .connect() .await?; - let mut trading_client = TradingServiceClient::new(channel.clone()); + let interceptor = create_auth_interceptor(token.clone()); + let mut trading_client = TradingServiceClient::with_interceptor(channel.clone(), interceptor); // Trading service should work (core functionality) let order_request = Request::new(SubmitOrderRequest { @@ -336,7 +354,8 @@ async fn test_e2e_trading_service_available_backtesting_optional() -> Result<()> println!("✓ Core trading service operational"); // Backtesting might be unavailable (graceful degradation) - let mut backtesting_client = BacktestingServiceClient::new(channel); + let interceptor2 = create_auth_interceptor(token); + let mut backtesting_client = BacktestingServiceClient::with_interceptor(channel, interceptor2); let backtest_request = Request::new(StartBacktestRequest { strategy_name: "test".to_string(), @@ -376,7 +395,8 @@ async fn test_e2e_partial_service_failure_handling() -> Result<()> { .connect() .await?; - let mut client = TradingServiceClient::new(channel); + let interceptor = create_auth_interceptor(token); + let mut client = TradingServiceClient::with_interceptor(channel, interceptor); // Check which services are available let status_request = Request::new(GetSystemStatusRequest { @@ -420,7 +440,8 @@ async fn test_e2e_circuit_breaker_validation() -> Result<()> { .connect() .await?; - let mut client = TradingServiceClient::new(channel); + let interceptor = create_auth_interceptor(token); + let mut client = TradingServiceClient::with_interceptor(channel, interceptor); // Submit multiple rapid requests to potentially trigger circuit breaker let mut success_count = 0; @@ -479,7 +500,8 @@ async fn test_e2e_timeout_handling() -> Result<()> { .connect() .await?; - let mut client = TradingServiceClient::new(channel); + let interceptor = create_auth_interceptor(token); + let mut client = TradingServiceClient::with_interceptor(channel, interceptor); let request = Request::new(trading::GetPositionsRequest { symbol: None, @@ -521,7 +543,8 @@ async fn test_e2e_retry_logic_validation() -> Result<()> { .connect() .await?; - let mut client = TradingServiceClient::new(channel); + let interceptor = create_auth_interceptor(token); + let mut client = TradingServiceClient::with_interceptor(channel, interceptor); // Simulate retryable operation let mut retry_count = 0; @@ -589,7 +612,8 @@ async fn test_e2e_api_gateway_routing() -> Result<()> { .connect() .await?; - let mut client = TradingServiceClient::new(channel); + let interceptor = create_auth_interceptor(token); + let mut client = TradingServiceClient::with_interceptor(channel, interceptor); // Test multiple routing paths let tests = vec![ @@ -627,7 +651,8 @@ async fn test_e2e_service_discovery() -> Result<()> { .connect() .await?; - let mut client = TradingServiceClient::new(channel); + let interceptor = create_auth_interceptor(token); + let mut client = TradingServiceClient::with_interceptor(channel, interceptor); let request = Request::new(GetSystemStatusRequest { service_names: vec![], @@ -674,7 +699,8 @@ async fn test_e2e_concurrent_service_requests() -> Result<()> { .await .unwrap(); - let mut client = TradingServiceClient::new(channel); + let interceptor = create_auth_interceptor(token_clone); + let mut client = TradingServiceClient::with_interceptor(channel, interceptor); let request = Request::new(trading::GetAccountInfoRequest { account_id: format!("test_account_{}", i), @@ -718,7 +744,8 @@ async fn test_e2e_load_balancing_verification() -> Result<()> { .connect() .await?; - let mut client = TradingServiceClient::new(channel); + let interceptor = create_auth_interceptor(token); + let mut client = TradingServiceClient::with_interceptor(channel, interceptor); // Send multiple requests and track response times let mut response_times = vec![]; @@ -767,7 +794,8 @@ async fn test_e2e_service_failover() -> Result<()> { .connect() .await?; - let mut client = TradingServiceClient::new(channel); + let interceptor = create_auth_interceptor(token); + let mut client = TradingServiceClient::with_interceptor(channel, interceptor); // Monitor system status for failover indicators let request = Request::new(GetSystemStatusRequest { diff --git a/services/integration_tests/tests/trading_service_e2e.rs b/services/integration_tests/tests/trading_service_e2e.rs index a243d3b9d..9b1d7d0d8 100644 --- a/services/integration_tests/tests/trading_service_e2e.rs +++ b/services/integration_tests/tests/trading_service_e2e.rs @@ -68,7 +68,7 @@ fn generate_test_token(user_id: &str, roles: Vec, permissions: Vec Result> { +async fn create_authenticated_client() -> Result) -> Result, Status> + Clone>>> { let token = generate_test_token( "test_trader_001", vec!["trader".to_string()], @@ -83,10 +83,17 @@ async fn create_authenticated_client() -> Result> .connect() .await?; - let mut client = TradingServiceClient::new(channel); + // Create interceptor that injects JWT token into request metadata + let interceptor = move |mut req: Request<()>| -> Result, Status> { + let token_value = format!("Bearer {}", token); + let metadata_value = MetadataValue::try_from(token_value) + .map_err(|_| Status::internal("Failed to create metadata value"))?; - // Attach JWT token to all requests - let token_metadata = MetadataValue::try_from(format!("Bearer {}", token))?; + req.metadata_mut().insert("authorization", metadata_value); + Ok(req) + }; + + let client = TradingServiceClient::with_interceptor(channel, interceptor); Ok(client) } diff --git a/services/trading_service/src/metrics_server.rs b/services/trading_service/src/metrics_server.rs index e76bed234..85876843a 100644 --- a/services/trading_service/src/metrics_server.rs +++ b/services/trading_service/src/metrics_server.rs @@ -12,7 +12,6 @@ use axum::{ routing::get, Router, }; -use std::sync::Arc; use std::time::Duration; use tokio::net::TcpListener; use tracing::{error, info, warn}; diff --git a/services/trading_service/src/repository_impls.rs b/services/trading_service/src/repository_impls.rs index c9301a06c..3754e5312 100644 --- a/services/trading_service/src/repository_impls.rs +++ b/services/trading_service/src/repository_impls.rs @@ -38,21 +38,70 @@ impl TradingRepository for PostgresTradingRepository { let order_id = uuid::Uuid::new_v4().to_string(); // Use direct sqlx query binding instead of trait objects + // Convert float prices to BIGINT (cents) - multiply by 100 for cent precision + // Market orders must have NULL limit_price per database constraint + let limit_price_cents = if order.order_type == common::OrderType::Market { + None + } else { + Some((order.price * 100.0) as i64) + }; + let stop_price_cents = order.stop_price.map(|p| (p * 100.0) as i64); + + // Convert quantity to BIGINT (base units) + let quantity_bigint = (order.quantity * 100.0) as i64; + + // Use nanosecond timestamp directly + let timestamp_ns = order.timestamp * 1_000_000_000; // Convert seconds to nanoseconds + + // Convert enums to PostgreSQL enum strings + let side_str = match order.side { + common::OrderSide::Buy => "buy", + common::OrderSide::Sell => "sell", + }; + + let order_type_str = match order.order_type { + common::OrderType::Market => "market", + common::OrderType::Limit => "limit", + common::OrderType::Stop => "stop", + common::OrderType::StopLimit => "stop_limit", + _ => "market", // Default fallback + }; + + let status_str = match order.status { + common::OrderStatus::Pending => "pending", + common::OrderStatus::New => "accepted", + common::OrderStatus::Rejected => "rejected", + common::OrderStatus::PartiallyFilled => "partial", + common::OrderStatus::Filled => "filled", + common::OrderStatus::Cancelled => "cancelled", + common::OrderStatus::Expired => "expired", + common::OrderStatus::Submitted => "submitted", + common::OrderStatus::Created => "created", + common::OrderStatus::Working => "working", + common::OrderStatus::Unknown => "unknown", + common::OrderStatus::Suspended => "suspended", + common::OrderStatus::PendingCancel => "pending_cancel", + common::OrderStatus::PendingReplace => "pending_replace", + _ => "unknown", // Catch-all for non-exhaustive enum + }; + let query = r#" - INSERT INTO orders (id, account_id, symbol, side, order_type, quantity, price, status, timestamp) - VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9) + INSERT INTO orders (id, account_id, symbol, side, order_type, quantity, limit_price, stop_price, status, created_at, updated_at, venue) + VALUES ($1, $2, $3, $4::order_side, $5::order_type, $6, $7, $8, $9::order_status, $10, $10, $11) "#; sqlx::query(query) .bind(&order_id) .bind(&order.account_id) .bind(&order.symbol) - .bind(order.side as i32) - .bind(order.order_type as i32) - .bind(order.quantity) - .bind(order.price) - .bind(order.status as i32) - .bind(safe_timestamp_to_datetime(order.timestamp)?) + .bind(side_str) + .bind(order_type_str) + .bind(quantity_bigint) + .bind(limit_price_cents) + .bind(stop_price_cents) + .bind(status_str) + .bind(timestamp_ns) + .bind("SIMULATED") // venue - required field .execute(&self.pool) .await .map_err(|e| TradingServiceError::DatabaseError { @@ -67,10 +116,30 @@ impl TradingRepository for PostgresTradingRepository { order_id: &str, status: common::types::OrderStatus, ) -> TradingServiceResult<()> { - let query = "UPDATE orders SET status = $1, updated_at = NOW() WHERE id = $2"; + let status_str = match status { + common::OrderStatus::Pending => "pending", + common::OrderStatus::New => "accepted", + common::OrderStatus::Rejected => "rejected", + common::OrderStatus::PartiallyFilled => "partial", + common::OrderStatus::Filled => "filled", + common::OrderStatus::Cancelled => "cancelled", + common::OrderStatus::Expired => "expired", + common::OrderStatus::Submitted => "submitted", + common::OrderStatus::Created => "created", + common::OrderStatus::Working => "working", + common::OrderStatus::Unknown => "unknown", + common::OrderStatus::Suspended => "suspended", + common::OrderStatus::PendingCancel => "pending_cancel", + common::OrderStatus::PendingReplace => "pending_replace", + _ => "unknown", // Catch-all for non-exhaustive enum + }; + + let timestamp_ns = chrono::Utc::now().timestamp() * 1_000_000_000; + let query = "UPDATE orders SET status = $1::order_status, updated_at = $2 WHERE id = $3"; sqlx::query(query) - .bind(status as i32) + .bind(status_str) + .bind(timestamp_ns) .bind(order_id) .execute(&self.pool) .await @@ -82,7 +151,7 @@ impl TradingRepository for PostgresTradingRepository { } async fn get_order(&self, order_id: &str) -> TradingServiceResult> { - let query = "SELECT id, account_id, symbol, side, order_type, quantity, price, status, EXTRACT(EPOCH FROM timestamp)::bigint as timestamp FROM orders WHERE id = $1"; + let query = "SELECT id, account_id, symbol, side, order_type, quantity, limit_price, stop_price, filled_quantity, status, created_at FROM orders WHERE id = $1"; let row = sqlx::query(query) .bind(order_id) @@ -93,21 +162,62 @@ impl TradingRepository for PostgresTradingRepository { })?; if let Some(row) = row { + // Convert BIGINT back to f64 + let quantity_bigint: i64 = row.get("quantity"); + let limit_price_bigint: Option = row.get("limit_price"); + let stop_price_bigint: Option = row.get("stop_price"); + let filled_quantity_bigint: i64 = row.get("filled_quantity"); + let created_at_ns: i64 = row.get("created_at"); + + // Parse enum strings from PostgreSQL + let side_str: String = row.get("side"); + let order_type_str: String = row.get("order_type"); + let status_str: String = row.get("status"); + + let side = match side_str.as_str() { + "buy" => common::OrderSide::Buy, + "sell" => common::OrderSide::Sell, + _ => common::OrderSide::Buy, + }; + + let order_type = match order_type_str.as_str() { + "market" => common::OrderType::Market, + "limit" => common::OrderType::Limit, + "stop" => common::OrderType::Stop, + "stop_limit" => common::OrderType::StopLimit, + _ => common::OrderType::Market, + }; + + let status = match status_str.as_str() { + "pending" => common::OrderStatus::Pending, + "accepted" => common::OrderStatus::New, + "rejected" => common::OrderStatus::Rejected, + "partial" => common::OrderStatus::PartiallyFilled, + "filled" => common::OrderStatus::Filled, + "cancelled" => common::OrderStatus::Cancelled, + "expired" => common::OrderStatus::Expired, + "submitted" => common::OrderStatus::Submitted, + "created" => common::OrderStatus::Created, + "working" => common::OrderStatus::Working, + "unknown" => common::OrderStatus::Unknown, + "suspended" => common::OrderStatus::Suspended, + "pending_cancel" => common::OrderStatus::PendingCancel, + "pending_replace" => common::OrderStatus::PendingReplace, + _ => common::OrderStatus::Pending, + }; + Ok(Some(TradingOrder { id: row.get("id"), account_id: row.get("account_id"), symbol: row.get("symbol"), - side: common::OrderSide::try_from(row.get::("side")) - .unwrap_or(common::OrderSide::Buy), - order_type: common::OrderType::try_from(row.get::("order_type")) - .unwrap_or(common::OrderType::Market), - quantity: row.get("quantity"), - filled_quantity: row.get::, _>("filled_quantity").unwrap_or(0.0), - price: row.get("price"), - stop_price: row.get::, _>("stop_price"), - status: common::OrderStatus::try_from(row.get::("status")) - .unwrap_or(common::OrderStatus::Pending), - timestamp: row.get::, _>("timestamp").unwrap_or(0), + side, + order_type, + quantity: quantity_bigint as f64 / 100.0, + filled_quantity: filled_quantity_bigint as f64 / 100.0, + price: limit_price_bigint.map(|p| p as f64 / 100.0).unwrap_or(0.0), + stop_price: stop_price_bigint.map(|p| p as f64 / 100.0), + status, + timestamp: created_at_ns / 1_000_000_000, // Convert nanoseconds to seconds })) } else { Ok(None) @@ -119,7 +229,7 @@ impl TradingRepository for PostgresTradingRepository { account_id: &str, ) -> TradingServiceResult> { let rows = sqlx::query( - "SELECT id, account_id, symbol, side, order_type, quantity, price, status, EXTRACT(EPOCH FROM timestamp)::bigint as timestamp FROM orders WHERE account_id = $1 ORDER BY timestamp DESC" + "SELECT id, account_id, symbol, side, order_type, quantity, limit_price, stop_price, filled_quantity, status, created_at FROM orders WHERE account_id = $1 ORDER BY created_at DESC" ) .bind(account_id) .fetch_all(&self.pool) @@ -128,21 +238,64 @@ impl TradingRepository for PostgresTradingRepository { let orders = rows .into_iter() - .map(|row| TradingOrder { - id: row.get("id"), - account_id: row.get("account_id"), - symbol: row.get("symbol"), - side: common::OrderSide::try_from(row.get::("side")) - .unwrap_or(common::OrderSide::Buy), - order_type: common::OrderType::try_from(row.get::("order_type")) - .unwrap_or(common::OrderType::Market), - quantity: row.get("quantity"), - filled_quantity: row.get::, _>("filled_quantity").unwrap_or(0.0), - price: row.get("price"), - stop_price: row.get::, _>("stop_price"), - status: common::OrderStatus::try_from(row.get::("status")) - .unwrap_or(common::OrderStatus::Pending), - timestamp: row.get::, _>("timestamp").unwrap_or(0), + .map(|row| { + // Convert BIGINT back to f64 + let quantity_bigint: i64 = row.get("quantity"); + let limit_price_bigint: Option = row.get("limit_price"); + let stop_price_bigint: Option = row.get("stop_price"); + let filled_quantity_bigint: i64 = row.get("filled_quantity"); + let created_at_ns: i64 = row.get("created_at"); + + // Parse enum strings from PostgreSQL + let side_str: String = row.get("side"); + let order_type_str: String = row.get("order_type"); + let status_str: String = row.get("status"); + + let side = match side_str.as_str() { + "buy" => common::OrderSide::Buy, + "sell" => common::OrderSide::Sell, + _ => common::OrderSide::Buy, + }; + + let order_type = match order_type_str.as_str() { + "market" => common::OrderType::Market, + "limit" => common::OrderType::Limit, + "stop" => common::OrderType::Stop, + "stop_limit" => common::OrderType::StopLimit, + _ => common::OrderType::Market, + }; + + let status = match status_str.as_str() { + "pending" => common::OrderStatus::Pending, + "accepted" => common::OrderStatus::New, + "rejected" => common::OrderStatus::Rejected, + "partial" => common::OrderStatus::PartiallyFilled, + "filled" => common::OrderStatus::Filled, + "cancelled" => common::OrderStatus::Cancelled, + "expired" => common::OrderStatus::Expired, + "submitted" => common::OrderStatus::Submitted, + "created" => common::OrderStatus::Created, + "working" => common::OrderStatus::Working, + "unknown" => common::OrderStatus::Unknown, + "suspended" => common::OrderStatus::Suspended, + "pending_cancel" => common::OrderStatus::PendingCancel, + "pending_replace" => common::OrderStatus::PendingReplace, + _ => common::OrderStatus::Pending, + }; + + TradingOrder { + id: row.get("id"), + account_id: row.get("account_id"), + symbol: row.get("symbol"), + side, + order_type, + quantity: quantity_bigint as f64 / 100.0, + filled_quantity: filled_quantity_bigint as f64 / 100.0, + price: limit_price_bigint.map(|p| p as f64 / 100.0).unwrap_or(0.0), + stop_price: stop_price_bigint.map(|p| p as f64 / 100.0), + status, + timestamp: created_at_ns / 1_000_000_000, + } }) .collect();