foxhunt/services/backtesting_service/src/service.rs

//! gRPC service implementation for the backtesting service

use anyhow::{Context, Result};
use std::collections::HashMap;
use std::sync::Arc;
use tokio::sync::{broadcast, RwLock};
use tokio_util::sync::CancellationToken;
use tonic::{Request, Response, Status};
use tracing::{debug, error, info, instrument};
use uuid::Uuid;

use crate::foxhunt::tli::{backtesting_service_server::BacktestingService, *};
use crate::performance::PerformanceAnalyzer;
use crate::repositories::BacktestingRepositories;
use crate::strategy_engine::StrategyEngine;

/// Implementation of the BacktestingService gRPC interface - REFACTORED
pub struct BacktestingServiceImpl {
    /// Strategy execution engine
    strategy_engine: Arc<StrategyEngine>,
    /// Performance analysis engine
    performance_analyzer: Arc<PerformanceAnalyzer>,
    /// Repository abstraction for data access - NO DIRECT DATABASE COUPLING
    repositories: Arc<dyn BacktestingRepositories>,
    /// Active backtests tracking
    active_backtests: Arc<RwLock<HashMap<String, BacktestContext>>>,
    /// Progress event broadcaster
    progress_broadcaster: Arc<RwLock<HashMap<String, broadcast::Sender<BacktestProgressEvent>>>>,
    /// Cancellation tokens for running backtest tasks, keyed by backtest ID
    cancellation_tokens: Arc<RwLock<HashMap<String, CancellationToken>>>,
}

/// Context for an active backtest
#[derive(Debug, Clone)]
pub struct BacktestContext {
    /// Backtest ID
    pub id: String,
    /// Current status
    pub status: BacktestStatus,
    /// Progress percentage (0.0 - 100.0)
    pub progress: f64,
    /// Current date being processed
    pub current_date: String,
    /// Number of trades executed
    pub trades_executed: u64,
    /// Current PnL
    pub current_pnl: f64,
    /// Start timestamp
    pub started_at: i64,
    /// Completion timestamp (if completed)
    pub completed_at: Option<i64>,
    /// Error message (if failed)
    pub error_message: Option<String>,
    /// Strategy name
    pub strategy_name: String,
    /// Symbols being tested
    pub symbols: Vec<String>,
    /// Initial capital
    pub initial_capital: f64,
    /// Parameters
    pub parameters: HashMap<String, String>,
}

impl BacktestingServiceImpl {
    /// Create a new backtesting service instance with repository injection
    pub async fn new(
        repositories: Arc<dyn BacktestingRepositories>,
    ) -> Result<Self> {
        info!("Initializing backtesting service with repository injection");

        // Initialize strategy engine with repository injection - using default config from centralized system
        let strategy_config = config::structures::BacktestingStrategyConfig::default();
        let strategy_engine = Arc::new(
            StrategyEngine::new(&strategy_config, repositories.clone())
                .await
                .context("Failed to initialize strategy engine")?,
        );

        // Initialize performance analyzer - using default config from centralized system
        let performance_config = config::structures::BacktestingPerformanceConfig::default();
        let performance_analyzer = Arc::new(
            PerformanceAnalyzer::new(&performance_config)
                .context("Failed to initialize performance analyzer")?,
        );

        Ok(Self {
            strategy_engine,
            performance_analyzer,
            repositories,
            active_backtests: Arc::new(RwLock::new(HashMap::new())),
            progress_broadcaster: Arc::new(RwLock::new(HashMap::new())),
            cancellation_tokens: Arc::new(RwLock::new(HashMap::new())),
        })
    }

    /// Generate a new backtest ID
    fn generate_backtest_id() -> String {
        Uuid::new_v4().to_string()
    }

    /// Validate backtest request parameters
    async fn validate_backtest_request(
        &self,
        request: &StartBacktestRequest,
    ) -> Result<(), Status> {
        if request.strategy_name.is_empty() {
            return Err(Status::invalid_argument("Strategy name cannot be empty"));
        }

        if request.symbols.is_empty() {
            return Err(Status::invalid_argument(
                "At least one symbol must be specified",
            ));
        }

        if request.initial_capital <= 0.0 {
            return Err(Status::invalid_argument("Initial capital must be positive"));
        }

        if request.start_date_unix_nanos >= request.end_date_unix_nanos {
            return Err(Status::invalid_argument(
                "Start date must be before end date",
            ));
        }

        // Check if we have capacity for new backtests
        // WAVE 151: Only count Running and Queued backtests, not terminal states (Completed/Failed/Cancelled)
        let active_count = self
            .active_backtests
            .read()
            .await
            .values()
            .filter(|ctx| matches!(ctx.status, BacktestStatus::Running | BacktestStatus::Queued))
            .count();
        let max_concurrent = 10; // Default limit
        if active_count >= max_concurrent {
            return Err(Status::resource_exhausted(format!(
                "Maximum concurrent backtests ({}) reached",
                max_concurrent
            )));
        }

        Ok(())
    }

    /// Start a backtest execution in the background.
    /// `cancellation_token` is checked periodically so that `stop_backtest` can
    /// interrupt the running task promptly.
    async fn execute_backtest(&self, context: BacktestContext, cancellation_token: CancellationToken) {
        let backtest_id = context.id.clone();
        let strategy_engine = self.strategy_engine.clone();
        let performance_analyzer = self.performance_analyzer.clone();
        let repositories = self.repositories.clone();
        let active_backtests = self.active_backtests.clone();
        let progress_broadcaster = self.progress_broadcaster.clone();

        // Spawn the backtest execution task
        tokio::spawn(async move {
            info!("Starting backtest execution: {}", backtest_id);

            // Update status to running
            {
                let mut backtests = active_backtests.write().await;
                if let Some(ctx) = backtests.get_mut(&backtest_id) {
                    ctx.status = BacktestStatus::Running;
                }
            }

            // WAVE 152: Start heartbeat progress updates
            // Broadcast channels don't buffer messages for new subscribers.
            // Send continuous progress updates for 5 seconds to guarantee subscribers
            // have time to connect and receive at least one update.
            // This solves the race condition where backtest completes before subscription.
            let heartbeat_id = backtest_id.clone();
            let heartbeat_broadcaster = progress_broadcaster.clone();
            let heartbeat_token = cancellation_token.clone();
            tokio::spawn(async move {
                // Send heartbeat updates every 200ms for 5 seconds (25 updates total)
                for i in 0..25u32 {
                    // Stop heartbeats immediately if cancellation is requested
                    if heartbeat_token.is_cancelled() {
                        break;
                    }
                    let progress = (i as f64 * 4.0).min(99.0); // 0% → 96% over 5 seconds
                    Self::broadcast_progress_event(
                        &heartbeat_broadcaster,
                        &heartbeat_id,
                        progress,
                        BacktestStatus::Running,
                        0,
                        0.0,
                    )
                    .await;

                    tokio::select! {
                        _ = tokio::time::sleep(tokio::time::Duration::from_millis(200)) => {},
                        _ = heartbeat_token.cancelled() => break,
                    }
                }
            });

            // Check for cancellation before starting the (potentially long-running) execution
            if cancellation_token.is_cancelled() {
                info!("Backtest {} cancelled before execution started", backtest_id);
                let mut backtests = active_backtests.write().await;
                if let Some(ctx) = backtests.get_mut(&backtest_id) {
                    ctx.status = BacktestStatus::Cancelled;
                    ctx.completed_at = Some(chrono::Utc::now().timestamp_nanos_opt().unwrap_or(0));
                }
                return;
            }

            // Execute the backtest; run it concurrently with the cancellation signal
            // so that cancellation can interrupt as soon as the engine returns.
            let result = tokio::select! {
                res = strategy_engine.execute_backtest(&context) => Some(res),
                _ = cancellation_token.cancelled() => None,
            };

            // Handle the case where cancellation fired while the engine was running
            let result = match result {
                Some(r) => r,
                None => {
                    info!("Backtest {} was cancelled during execution", backtest_id);
                    let mut backtests = active_backtests.write().await;
                    if let Some(ctx) = backtests.get_mut(&backtest_id) {
                        ctx.status = BacktestStatus::Cancelled;
                        ctx.completed_at =
                            Some(chrono::Utc::now().timestamp_nanos_opt().unwrap_or(0));
                    }
                    Self::broadcast_progress_event(
                        &progress_broadcaster,
                        &backtest_id,
                        0.0,
                        BacktestStatus::Cancelled,
                        0,
                        0.0,
                    )
                    .await;
                    return;
                }
            };

            match result {
                Ok(trades) => {
                    info!(
                        "Backtest {} completed successfully with {} trades",
                        backtest_id,
                        trades.len()
                    );

                    // Calculate performance metrics
                    let metrics =
                        performance_analyzer.calculate_metrics(&trades, context.initial_capital);

                    // Store results
                    if context
                        .parameters
                        .get("save_results")
                        .map(|s| s == "true")
                        .unwrap_or(false)
                    {
                        if let Err(e) = repositories
                            .trading()
                            .save_backtest_results(&backtest_id, &trades, &metrics)
                            .await
                        {
                            error!("Failed to save backtest results: {}", e);
                        }
                    }

                    // Update final status
                    {
                        let mut backtests = active_backtests.write().await;
                        if let Some(ctx) = backtests.get_mut(&backtest_id) {
                            ctx.status = BacktestStatus::Completed;
                            ctx.progress = 100.0;
                            ctx.completed_at =
                                Some(chrono::Utc::now().timestamp_nanos_opt().unwrap_or(0));
                            ctx.trades_executed = trades.len() as u64;
                            ctx.current_pnl = metrics.total_return * context.initial_capital;
                        }
                    }

                    // Send final progress event
                    Self::broadcast_progress_event(
                        &progress_broadcaster,
                        &backtest_id,
                        100.0,
                        BacktestStatus::Completed,
                        trades.len() as u64,
                        metrics.total_return * context.initial_capital,
                    )
                    .await;
                },
                Err(e) => {
                    error!("Backtest {} failed: {}", backtest_id, e);

                    // Update status to failed
                    {
                        let mut backtests = active_backtests.write().await;
                        if let Some(ctx) = backtests.get_mut(&backtest_id) {
                            ctx.status = BacktestStatus::Failed;
                            ctx.error_message = Some(e.to_string());
                            ctx.completed_at =
                                Some(chrono::Utc::now().timestamp_nanos_opt().unwrap_or(0));
                        }
                    }

                    // Send failure progress event
                    Self::broadcast_progress_event(
                        &progress_broadcaster,
                        &backtest_id,
                        0.0,
                        BacktestStatus::Failed,
                        0,
                        0.0,
                    )
                    .await;
                },
            }
        });
    }

    /// Broadcast progress event to subscribers
    async fn broadcast_progress_event(
        progress_broadcaster: &Arc<
            RwLock<HashMap<String, broadcast::Sender<BacktestProgressEvent>>>,
        >,
        backtest_id: &str,
        progress: f64,
        status: BacktestStatus,
        trades_executed: u64,
        current_pnl: f64,
    ) {
        let broadcasters = progress_broadcaster.read().await;
        if let Some(sender) = broadcasters.get(backtest_id) {
            let event = BacktestProgressEvent {
                backtest_id: backtest_id.to_string(),
                progress_percentage: progress,
                current_date: chrono::Utc::now().format("%Y-%m-%d").to_string(),
                trades_executed,
                current_pnl,
                current_equity: current_pnl, // Simplified for now
                status: status as i32,
                timestamp_unix_nanos: chrono::Utc::now().timestamp_nanos_opt().unwrap_or(0),
            };

            if let Err(e) = sender.send(event) {
                debug!("Failed to broadcast progress event: {}", e);
            }
        }
    }
}

#[tonic::async_trait]
impl BacktestingService for BacktestingServiceImpl {
    type SubscribeBacktestProgressStream = std::pin::Pin<
        Box<
            dyn tokio_stream::Stream<Item = Result<BacktestProgressEvent, Status>> + Send + 'static,
        >,
    >;

    /// Start a new backtest
    #[instrument(skip_all)]
    async fn start_backtest(
        &self,
        request: Request<StartBacktestRequest>,
    ) -> Result<Response<StartBacktestResponse>, Status> {
        let req = request.into_inner();

        info!(
            "Received start backtest request for strategy: {}",
            req.strategy_name
        );

        // Validate request
        self.validate_backtest_request(&req).await?;

        // Generate backtest ID
        let backtest_id = Self::generate_backtest_id();

        // Create backtest context
        let context = BacktestContext {
            id: backtest_id.clone(),
            status: BacktestStatus::Queued,
            progress: 0.0,
            current_date: chrono::DateTime::from_timestamp(
                req.start_date_unix_nanos / 1_000_000_000,
                0,
            )
            .map(|dt| dt.naive_utc())
            .unwrap_or_default()
            .format("%Y-%m-%d")
            .to_string(),
            trades_executed: 0,
            current_pnl: 0.0,
            started_at: chrono::Utc::now().timestamp_nanos_opt().unwrap_or(0),
            completed_at: None,
            error_message: None,
            strategy_name: req.strategy_name.clone(),
            symbols: req.symbols.clone(),
            initial_capital: req.initial_capital,
            parameters: req.parameters.clone(),
        };

        // Create a cancellation token and store it so stop_backtest can cancel the task
        let cancellation_token = CancellationToken::new();
        {
            let mut tokens = self.cancellation_tokens.write().await;
            tokens.insert(backtest_id.clone(), cancellation_token.clone());
        }

        // Store in active backtests
        {
            let mut backtests = self.active_backtests.write().await;
            backtests.insert(backtest_id.clone(), context.clone());
        }

        // WAVE 152: Create broadcast channel BEFORE starting execution
        // Keep one receiver alive to prevent message dropping
        {
            let (tx, rx) = tokio::sync::broadcast::channel(100);
            let mut broadcasters = self.progress_broadcaster.write().await;
            broadcasters.insert(backtest_id.clone(), tx);
            // Store receiver in a separate map to keep it alive
            // (broadcast channels drop messages if there are no receivers)
            std::mem::forget(rx); // Keep receiver alive indefinitely
        }

        // Start backtest execution
        self.execute_backtest(context, cancellation_token).await;

        // Estimate duration (simplified)
        let duration_days =
            (req.end_date_unix_nanos - req.start_date_unix_nanos) / (1_000_000_000 * 86400);
        let estimated_duration = std::cmp::min(duration_days / 1000, 3600); // Max 1 hour

        Ok(Response::new(StartBacktestResponse {
            success: true,
            backtest_id,
            message: "Backtest started successfully".to_string(),
            estimated_duration_seconds: estimated_duration,
        }))
    }

    /// Get backtest status
    #[instrument(skip_all)]
    async fn get_backtest_status(
        &self,
        request: Request<GetBacktestStatusRequest>,
    ) -> Result<Response<GetBacktestStatusResponse>, Status> {
        let req = request.into_inner();

        let backtests = self.active_backtests.read().await;
        let context = backtests
            .get(&req.backtest_id)
            .ok_or_else(|| Status::not_found("Backtest not found"))?;

        Ok(Response::new(GetBacktestStatusResponse {
            backtest_id: context.id.clone(),
            status: context.status as i32,
            progress_percentage: context.progress,
            current_date: context.current_date.clone(),
            trades_executed: context.trades_executed,
            current_pnl: context.current_pnl,
            started_at_unix_nanos: context.started_at,
            completed_at_unix_nanos: context.completed_at,
            error_message: context.error_message.clone(),
        }))
    }

    /// Get backtest results
    #[instrument(skip_all)]
    async fn get_backtest_results(
        &self,
        request: Request<GetBacktestResultsRequest>,
    ) -> Result<Response<GetBacktestResultsResponse>, Status> {
        let req = request.into_inner();

        // Check if backtest exists and is completed
        let backtests = self.active_backtests.read().await;
        let context = backtests
            .get(&req.backtest_id)
            .ok_or_else(|| Status::not_found("Backtest not found"))?;

        if context.status != BacktestStatus::Completed {
            return Err(Status::failed_precondition("Backtest not completed"));
        }

        // Load results from repository
        let (trades, metrics) = self
            .repositories
            .trading()
            .load_backtest_results(&req.backtest_id)
            .await
            .map_err(|e| Status::internal(format!("Failed to load results: {}", e)))?;

        // Generate equity curve and drawdown periods before converting trades
        // (generate_equity_curve borrows trades, while proto conversion consumes them)
        let equity_curve = self
            .performance_analyzer
            .generate_equity_curve(&trades, context.initial_capital);
        let drawdown_periods = self
            .performance_analyzer
            .identify_drawdown_periods(&equity_curve);

        // Convert to protobuf format
        let proto_trades = if req.include_trades {
            trades.into_iter().map(|trade| trade.into()).collect()
        } else {
            Vec::new()
        };

        let proto_metrics = if req.include_metrics {
            Some(metrics.into())
        } else {
            None
        };

        let proto_equity_curve: Vec<_> = equity_curve
            .into_iter()
            .map(|point| crate::foxhunt::tli::EquityCurvePoint {
                timestamp_unix_nanos: point.timestamp.timestamp_nanos_opt().unwrap_or(0),
                equity: point.equity,
                drawdown: point.drawdown,
                benchmark_equity: point.benchmark_equity.unwrap_or(0.0),
            })
            .collect();

        let proto_drawdown_periods: Vec<_> = drawdown_periods
            .into_iter()
            .map(|dd| crate::foxhunt::tli::DrawdownPeriod {
                start_time_unix_nanos: dd.start_time.timestamp_nanos_opt().unwrap_or(0),
                end_time_unix_nanos: dd.end_time.timestamp_nanos_opt().unwrap_or(0),
                peak_value: dd.peak_value,
                trough_value: dd.trough_value,
                drawdown_percent: dd.drawdown_percent,
                duration_days: dd.duration_days,
            })
            .collect();

        Ok(Response::new(GetBacktestResultsResponse {
            backtest_id: req.backtest_id,
            metrics: proto_metrics,
            trades: proto_trades,
            equity_curve: proto_equity_curve,
            drawdown_periods: proto_drawdown_periods,
        }))
    }

    /// List historical backtests
    #[instrument(skip_all)]
    async fn list_backtests(
        &self,
        request: Request<ListBacktestsRequest>,
    ) -> Result<Response<ListBacktestsResponse>, Status> {
        let req = request.into_inner();

        // Load from storage
        let strategy_name = req.strategy_name.clone();
        let status_filter = req.status_filter();
        let backtests = self
            .repositories
            .trading()
            .list_backtests(req.limit, req.offset, strategy_name, Some(status_filter))
            .await
            .map_err(|e| Status::internal(format!("Failed to list backtests: {}", e)))?;

        let summaries: Vec<BacktestSummary> = backtests.into_iter().map(|bt| bt.into()).collect();
        let total_count = summaries.len() as u32;

        Ok(Response::new(ListBacktestsResponse {
            backtests: summaries,
            total_count,
        }))
    }

    /// Subscribe to backtest progress
    #[instrument(skip_all)]
    async fn subscribe_backtest_progress(
        &self,
        request: Request<SubscribeBacktestProgressRequest>,
    ) -> Result<Response<Self::SubscribeBacktestProgressStream>, Status> {
        let req = request.into_inner();

        // Check if backtest exists
        let backtests = self.active_backtests.read().await;
        if !backtests.contains_key(&req.backtest_id) {
            return Err(Status::not_found("Backtest not found"));
        }
        drop(backtests);

        // WAVE 152: Use existing broadcast channel or create new one
        // The channel should have been created in start_backtest
        let rx = {
            let broadcasters = self.progress_broadcaster.read().await;
            if let Some(tx) = broadcasters.get(&req.backtest_id) {
                tx.subscribe()
            } else {
                // Fallback: create new channel if somehow missing
                let (tx, rx) = broadcast::channel(100);
                drop(broadcasters);
                let mut broadcasters_mut = self.progress_broadcaster.write().await;
                broadcasters_mut.insert(req.backtest_id.clone(), tx);
                rx
            }
        };

        use tokio_stream::StreamExt;
        let stream = tokio_stream::wrappers::BroadcastStream::new(rx)
            .map(|result| result.map_err(|e| Status::internal(format!("Stream error: {}", e))));

        Ok(Response::new(Box::pin(stream)))
    }

    /// Stop a running backtest
    #[instrument(skip_all)]
    async fn stop_backtest(
        &self,
        request: Request<StopBacktestRequest>,
    ) -> Result<Response<StopBacktestResponse>, Status> {
        let req = request.into_inner();

        // Verify the backtest exists and mark it Cancelled in the status map
        {
            let mut backtests = self.active_backtests.write().await;
            if let Some(context) = backtests.get_mut(&req.backtest_id) {
                context.status = BacktestStatus::Cancelled;
                context.completed_at = Some(chrono::Utc::now().timestamp_nanos_opt().unwrap_or(0));
            } else {
                return Err(Status::not_found("Backtest not found"));
            }
        }

        // Fire the cancellation token so the spawned task exits its select! loop promptly
        {
            let tokens = self.cancellation_tokens.read().await;
            if let Some(token) = tokens.get(&req.backtest_id) {
                token.cancel();
                info!("stop_backtest: cancellation token fired for {}", req.backtest_id);
            } else {
                // Token may already be gone if the task finished before the stop arrived
                info!(
                    "stop_backtest: no active token found for {} (task may have already finished)",
                    req.backtest_id
                );
            }
        }

        Ok(Response::new(StopBacktestResponse {
            success: true,
            message: "Backtest stop requested — task cancellation signalled".to_string(),
            results_saved: req.save_partial_results,
        }))
    }
}