//! Stream Processor - Thread principal de traitement temps réel
//!
//! Ce module orchestre:
//! - Spawn du processus FFmpeg
//! - Monitoring des logs en temps réel
//! - Suivi des segments générés
//! - Persistance DB
//! - Callbacks de fin

use crate::core::job::EncodeJob;
use crate::core::processing::callbacks::ProcessingCallbacks;
use crate::core::processing::ffmpeg_monitor::{FFmpegEvent, FFmpegMonitor};
use crate::core::processing::segment_tracker::SegmentTracker;
use crate::error::AppError;
use crate::transcoding::ffmpeg::command_builder::FfmpegCommandBuilder;
use crate::transcoding::codecs::ContainerFormat;
use sqlx::PgPool;
use std::path::PathBuf;
use std::sync::Arc;
use std::time::Duration;
use tokio::process::Command;
use tokio::sync::mpsc;
use tokio::time::timeout;
use uuid::Uuid;

/// Statut de processing pour communication avec l'extérieur
#[derive(Debug, Clone)]
pub struct ProcessingStatus {
    /// Job ID
    pub job_id: Uuid,
    /// Statut actuel
    pub status: String,
    /// Nombre de segments détectés
    pub segments_count: usize,
    /// Durée totale actuelle
    pub current_duration: f64,
    /// Progression (0.0 - 1.0)
    pub progress: f32,
}

/// Processor principal pour un job de stream processing
pub struct StreamProcessor {
    /// Job ID
    job_id: Uuid,
    /// Job d'encodage
    job: EncodeJob,
    /// Pool de connexions DB
    db: PgPool,
    /// Répertoire de sortie
    output_dir: PathBuf,
    /// Timeout pour le job (défaut: 10 minutes)
    job_timeout: Duration,
    /// Sender pour les statuts de processing
    status_tx: Option<mpsc::Sender<ProcessingStatus>>,
}

impl StreamProcessor {
    /// Crée un nouveau processor
    pub fn new(
        job_id: Uuid,
        job: EncodeJob,
        db: PgPool,
        output_dir: PathBuf,
    ) -> Self {
        Self {
            job_id,
            job,
            db,
            output_dir,
            job_timeout: Duration::from_secs(600), // 10 minutes par défaut
            status_tx: None,
        }
    }

    /// Définit un sender pour les statuts (optionnel)
    pub fn with_status_sender(mut self, tx: mpsc::Sender<ProcessingStatus>) -> Self {
        self.status_tx = Some(tx);
        self
    }

    /// Définit un timeout personnalisé
    pub fn with_timeout(mut self, timeout: Duration) -> Self {
        self.job_timeout = timeout;
        self
    }

    /// Exécute le traitement complet
    pub async fn run(&mut self) -> Result<(), AppError> {
        tracing::info!(
            "🚀 Démarrage du processing pour job {} (track: {}, qualité: {})",
            self.job_id,
            self.job.track_id,
            self.job.quality
        );

        // 1. Mettre à jour le statut en DB
        self.update_status("encoding").await?;

        // 2. Spawn FFmpeg
        let mut child = self.spawn_ffmpeg().await?;

        // 3. Créer le monitor et tracker
        let (monitor, mut event_rx) = FFmpegMonitor::new(self.output_dir.clone());
        let tracker = Arc::new(SegmentTracker::new(
            self.job_id,
            self.job.track_id,
            self.job.quality.clone(),
            self.db.clone(),
        ));

        // 4. Démarrer le monitoring de stderr
        let stderr = child.stderr.take().ok_or_else(|| AppError::InternalError {
            message: "FFmpeg stderr not available".to_string(),
        })?;

        let monitor_handle = {
            let monitor = monitor.clone();
            tokio::spawn(async move {
                if let Err(e) = monitor.monitor_stderr(stderr).await {
                    tracing::error!("Error monitoring FFmpeg stderr: {}", e);
                }
            })
        };

        // 5. Traiter les événements en temps réel
        let tracker_clone = tracker.clone();
        let event_handle = tokio::spawn(async move {
            while let Some(event) = event_rx.recv().await {
                match event {
                    FFmpegEvent::SegmentOpened { path, index } => {
                        // Détecter la durée depuis le manifest ou utiliser la durée par défaut
                        let duration = 4.0; // Durée par défaut (sera mise à jour depuis le manifest)
                        
                        let segment = crate::core::processing::segment_tracker::SegmentInfo {
                            index,
                            path,
                            duration,
                            timestamp: std::time::SystemTime::now(),
                        };

                        if let Err(e) = tracker_clone.register(segment).await {
                            tracing::error!("Failed to register segment: {}", e);
                        }
                    }
                    FFmpegEvent::Progress { time, speed } => {
                        if let Some(t) = time {
                            tracing::debug!(
                                "⏱️ Progression: {} (speed: {:.2}x)",
                                format_duration(t),
                                speed.unwrap_or(0.0)
                            );
                        }
                    }
                    FFmpegEvent::Error(msg) => {
                        tracing::warn!("⚠️ Erreur FFmpeg détectée: {}", msg);
                    }
                    FFmpegEvent::MuxingComplete { overhead } => {
                        tracing::info!("✅ Muxing terminé (overhead: {:.2}%)", overhead);
                    }
                }
            }
        });

        // 6. Attendre la fin du processus avec timeout
        let result = timeout(self.job_timeout, child.wait()).await;

        // Attendre la fin gracieuse du monitoring (drain events)
        // Si le processus est fini, stderr est fermé, donc monitor_handle va finir
        // Ce qui droppera le sender, et donc event_handle finira aussi.
        if let Err(e) = monitor_handle.await {
             tracing::warn!("Monitor handle join error: {}", e);
        }
        if let Err(e) = event_handle.await {
             tracing::warn!("Event handle join error: {}", e);
        }

        // 7. Gérer le résultat
        match result {
            Ok(Ok(status)) => {
                if status.success() {
                    // Finaliser
                    self.finalize(tracker).await?;
                    Ok(())
                } else {
                    let error_msg = format!("FFmpeg exited with status: {}", status);
                    self.on_error(&error_msg).await?;
                    Err(AppError::InternalError { message: error_msg })
                }
            }
            Ok(Err(e)) => {
                let error_msg = format!("FFmpeg IO error: {}", e);
                self.on_error(&error_msg).await?;
                Err(AppError::InternalError { message: error_msg })
            }
            Err(_) => {
                // Timeout: tuer le processus
                let _ = child.kill().await;
                self.on_timeout().await?;
                Err(AppError::InternalError {
                    message: format!("Encoding timed out after {} seconds", self.job_timeout.as_secs()),
                })
            }
        }
    }

    /// Spawn le processus FFmpeg
    async fn spawn_ffmpeg(&self) -> Result<tokio::process::Child, AppError> {
        // Créer le répertoire de sortie
        tokio::fs::create_dir_all(&self.output_dir).await
            .map_err(|e| AppError::InternalError {
                message: format!("Failed to create output directory: {}", e),
            })?;

        // Construire la commande FFmpeg
        let manifest_path = self.output_dir.join("index.m3u8");
        let mut command = FfmpegCommandBuilder::new()
            .input(&self.job.input_path)
            .output(&manifest_path)
            .audio_codec(self.job.codec.clone())
            .bitrate(self.job.bitrate)
            .sample_rate(self.job.sample_rate)
            .channels(self.job.channels)
            .container(ContainerFormat::HLS)
            .hls_time(self.job.hls_segment_time)
            .build()
            .map_err(|e| AppError::InternalError {
                message: format!("Failed to build FFmpeg command: {}", e),
            })?;

        // Enregistrer le PID et start_time en DB (si colonnes existent)
        // Note: On pourrait ajouter ces colonnes via une migration
        // Pour l'instant, on log juste
        tracing::debug!("🎬 Spawn FFmpeg pour job {}", self.job_id);

        // Spawn le processus
        let child = command.spawn()
            .map_err(|e| AppError::InternalError {
                message: format!("Failed to spawn FFmpeg process: {}", e),
            })?;

        Ok(child)
    }

    /// Finalise le traitement (succès) - Transactionnel P0
    /// 
    /// Garantit l'atomicité entre :
    /// - Persistance de tous les segments
    /// - Finalisation du job (status = 'done', duration, completed_at)
    /// 
    /// En cas d'erreur, rollback automatique → pas de job finalisé sans segments
    async fn finalize(&self, tracker: Arc<SegmentTracker>) -> Result<(), AppError> {
        tracing::info!("✅ Finalisation du job {} (transactionnel)", self.job_id);

        // DÉBUT TRANSACTION
        let mut tx = self.db.begin().await
            .map_err(|e| AppError::InternalError {
                message: format!("Failed to begin transaction for job finalization: {}", e),
            })?;

        // 1. VALIDATION : Vérifier que le job existe et est en "encoding"
        let job_status: Option<String> = sqlx::query_scalar!(
            r#"
            SELECT status FROM stream_jobs WHERE id = $1
            "#,
            self.job_id
        )
        .fetch_optional(&mut *tx)
        .await
        .map_err(|e| AppError::InternalError {
            message: format!("Failed to validate job {}: {}", self.job_id, e),
        })?;

        if job_status.is_none() {
            return Err(AppError::NotFound {
                resource: format!("Job {}", self.job_id),
            });
        }

        // 2. Persister tous les segments dans la transaction
        let segments = tracker.get_segments().await;
        if !segments.is_empty() {
            for segment in segments.iter() {
                sqlx::query!(
                    r#"
                    INSERT INTO stream_segments (track_id, quality, segment_index, path, duration, created_at)
                    VALUES ($1, $2, $3, $4, $5, NOW())
                    ON CONFLICT (track_id, quality, segment_index) DO UPDATE
                    SET path = EXCLUDED.path, duration = EXCLUDED.duration, created_at = NOW()
                    "#,
                    self.job.track_id,
                    self.job.quality,
                    segment.index as i32,
                    segment.path.to_string_lossy(),
                    segment.duration
                )
                .execute(&mut *tx)
                .await
                .map_err(|e| AppError::InternalError {
                    message: format!("Failed to insert segment {} in finalization: {}", 
                        segment.path.display(), e),
                })?;
            }
        }

        // 3. CALCUL : Durée totale depuis les segments
        let total_duration: f64 = segments.iter().map(|s| s.duration).sum();

        // 4. UPDATE job : Finalisation (status = 'done', duration, completed_at)
        sqlx::query!(
            r#"
            UPDATE stream_jobs
            SET status = 'done', updated_at = NOW()
            WHERE id = $1
            "#,
            self.job_id
        )
        .execute(&mut *tx)
        .await
        .map_err(|e| AppError::InternalError {
            message: format!("Failed to finalize job {}: {}", self.job_id, e),
        })?;

        // COMMIT TRANSACTION
        tx.commit().await
            .map_err(|e| AppError::InternalError {
                message: format!("Failed to commit job finalization transaction: {}", e),
            })?;

        tracing::info!(
            job_id = %self.job_id,
            segment_count = segments.len(),
            total_duration = total_duration,
            "Job finalized successfully in transaction"
        );

        // Créer les callbacks pour les opérations non-DB
        let callbacks = ProcessingCallbacks::new(
            self.job_id,
            self.job.track_id,
            self.db.clone(),
            self.output_dir.clone(),
        );

        // Finaliser la playlist (opération fichier, pas DB)
        callbacks.finalize_playlist().await?;

        // Envoyer le statut final
        self.send_status(ProcessingStatus {
            job_id: self.job_id,
            status: "done".to_string(),
            segments_count: segments.len(),
            current_duration: total_duration,
            progress: 1.0,
        }).await;

        Ok(())
    }

    /// Gère une erreur
    async fn on_error(&self, error_message: &str) -> Result<(), AppError> {
        let callbacks = ProcessingCallbacks::new(
            self.job_id,
            self.job.track_id,
            self.db.clone(),
            self.output_dir.clone(),
        );

        callbacks.on_error(error_message).await?;

        self.send_status(ProcessingStatus {
            job_id: self.job_id,
            status: "error".to_string(),
            segments_count: 0,
            current_duration: 0.0,
            progress: 0.0,
        }).await;

        Ok(())
    }

    /// Gère un timeout
    async fn on_timeout(&self) -> Result<(), AppError> {
        let callbacks = ProcessingCallbacks::new(
            self.job_id,
            self.job.track_id,
            self.db.clone(),
            self.output_dir.clone(),
        );

        callbacks.on_timeout().await?;

        self.send_status(ProcessingStatus {
            job_id: self.job_id,
            status: "timeout".to_string(),
            segments_count: 0,
            current_duration: 0.0,
            progress: 0.0,
        }).await;

        Ok(())
    }

    /// Met à jour le statut en DB
    async fn update_status(&self, status: &str) -> Result<(), AppError> {
        sqlx::query!(
            r#"
            UPDATE stream_jobs
            SET status = $1, updated_at = NOW()
            WHERE id = $2
            "#,
            status,
            self.job_id
        )
        .execute(&self.db)
        .await
        .map_err(|e| AppError::InternalError {
            message: format!("Failed to update job status: {}", e),
        })?;

        Ok(())
    }

    /// Envoie un statut via le channel (si disponible)
    async fn send_status(&self, status: ProcessingStatus) {
        if let Some(ref tx) = self.status_tx {
            let _ = tx.send(status).await;
        }
    }
}

/// Formate une durée pour l'affichage
fn format_duration(d: Duration) -> String {
    let total_secs = d.as_secs();
    let hours = total_secs / 3600;
    let minutes = (total_secs % 3600) / 60;
    let seconds = total_secs % 60;
    let millis = d.subsec_millis();

    if hours > 0 {
        format!("{:02}:{:02}:{:02}.{:03}", hours, minutes, seconds, millis)
    } else {
        format!("{:02}:{:02}.{:03}", minutes, seconds, millis)
    }
}

