//! Pool d'encodeurs avec workers FFmpeg
//!
//! Ce module implémente un pool de workers qui exécutent des jobs d'encodage
//! audio via FFmpeg. Chaque worker est un thread dédié qui spawn un processus
//! FFmpeg et capture sa sortie pour monitoring.

use crate::core::job::{EncodeJob, EncodeJobStatus};
use crate::error::AppError;
use crate::transcoding::ffmpeg::command_builder::FfmpegCommandBuilder;
use crate::transcoding::ffmpeg::progress_parser::FfmpegProgress;
use crate::transcoding::codecs::ContainerFormat;
use async_channel::{Receiver, Sender};
use sqlx::PgPool;
use std::time::Duration;
use tokio::time::timeout;
use uuid::Uuid;
// Added imports for stubs
use std::sync::Arc;
use crate::core::stream::{StreamSource, StreamOutput};

/// Pool d'encodeurs avec workers FFmpeg
#[derive(Clone, Debug)]
pub struct EncoderPool {
    /// Sender pour envoyer des jobs dans la queue
    sender: Sender<EncodeJob>,
    /// Pool de connexions DB
    db_pool: PgPool,
    /// Nombre de workers actifs
    worker_count: usize,
}

/// Pipeline d'encodage (Stub pour compilation)
#[derive(Debug)]
pub struct EncoderPipeline;

/// Worker d'encodage qui traite les jobs
struct EncoderWorker {
    id: usize,
    receiver: Receiver<EncodeJob>,
    db_pool: PgPool,
}

/// Timeout maximum pour un job d'encodage (10 minutes)
const JOB_TIMEOUT: Duration = Duration::from_secs(600);

impl EncoderPool {
    /// Crée un nouveau pool d'encodeurs
    ///
    /// # Arguments
    ///
    /// * `db_pool` - Pool de connexions PostgreSQL
    /// * `worker_count` - Nombre de workers (par défaut: min(nb_cpu / 2, 8))
    ///
    /// # Returns
    ///
    /// Un `EncoderPool` configuré avec les workers démarrés
    pub async fn new(db_pool: PgPool, worker_count: Option<usize>) -> Result<Self, AppError> {
        let worker_count = worker_count.unwrap_or_else(|| {
            let cpu_count = num_cpus::get();
            std::cmp::min(cpu_count / 2, 8).max(1)
        });

        let (sender, receiver) = async_channel::unbounded();

        let pool = Self {
            sender,
            db_pool: db_pool.clone(),
            worker_count,
        };

        // Démarrer les workers
        for i in 0..worker_count {
            let worker = EncoderWorker {
                id: i,
                receiver: receiver.clone(),
                db_pool: db_pool.clone(),
            };
            tokio::spawn(async move {
                worker.run().await;
            });
            tracing::info!("Encoder worker {} démarré", i);
        }

        tracing::info!(
            "EncoderPool créé avec {} workers",
            worker_count
        );

        Ok(pool)
    }

    /// Envoie un job d'encodage dans la queue
    ///
    /// # Arguments
    ///
    /// * `job` - Job d'encodage à traiter
    ///
    /// # Returns
    ///
    /// `Ok(())` si le job a été envoyé, erreur sinon
    pub async fn submit_job(&self, job: EncodeJob) -> Result<(), AppError> {
        // Créer l'entrée en DB
        let job_id = Uuid::new_v4();
        sqlx::query(
            r#"
            INSERT INTO stream_jobs (id, track_id, status)
            VALUES ($1, $2, 'pending')
            "#,
        )
        .bind(job_id)
        .bind(job.track_id)
        .execute(&self.db_pool)
        .await
        .map_err(|e| AppError::InternalError {
            message: format!("Failed to create stream_job: {}", e),
        })?;

        // Envoyer dans la queue
        self.sender
            .send(job.clone())
            .await
            .map_err(|_| AppError::InternalError {
                message: "Failed to submit job to queue".to_string(),
            })?;

        tracing::info!("Job {} soumis pour track {}", job_id, job.track_id);
        Ok(())
    }

    /// Obtient le nombre de workers
    pub fn worker_count(&self) -> usize {
        self.worker_count
    }

    /// Crée un pipeline d'encodage (Stub)
    pub async fn create_pipeline(
        &self,
        _source: &StreamSource,
        _outputs: &[StreamOutput], // Use slice reference instead of Vec for flexibility
    ) -> Result<Arc<EncoderPipeline>, AppError> {
        // TODO: Implémenter la création réelle du pipeline
        tracing::debug!("Creating stub encoder pipeline");
        Ok(Arc::new(EncoderPipeline))
    }
}

impl EncoderWorker {
    /// Boucle principale du worker
    async fn run(self) {
        tracing::info!("EncoderWorker {} démarré", self.id);

        loop {
            // Recevoir un job (bloquant)
            match self.receiver.recv().await {
                Ok(job) => {
                    tracing::info!(
                        "Worker {} traite job pour track {} (qualité: {})",
                        self.id,
                        job.track_id,
                        job.quality
                    );

                    // Mettre à jour le statut en DB
                    if let Err(e) = self.update_job_status(&job.track_id, EncodeJobStatus::Encoding).await {
                        tracing::error!("Worker {} failed to update status: {}", self.id, e);
                        continue;
                    }

                    // Traiter le job
                    match self.process_job(job).await {
                        Ok(_) => {
                            tracing::info!("Worker {} a terminé le job avec succès", self.id);
                        }
                        Err(e) => {
                            tracing::error!("Worker {} job failed: {}", self.id, e);
                        }
                    }
                }
                Err(_) => {
                    // Channel fermé, arrêter le worker
                    tracing::info!("Worker {} arrêté (channel fermé)", self.id);
                    break;
                }
            }
        }
    }

    /// Traite un job d'encodage
    async fn process_job(&self, job: EncodeJob) -> Result<(), AppError> {
        // 1. Créer le répertoire de sortie
        tokio::fs::create_dir_all(&job.output_dir)
            .await
            .map_err(|e| AppError::InternalError {
                message: format!("Failed to create output directory: {}", e),
            })?;

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

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

        // 4. Capturer stderr en streaming pour monitoring
        let stderr = child.stderr.take();
        if let Some(mut stderr) = stderr {
            let worker_id = self.id;
            let track_id = job.track_id;
            tokio::spawn(async move {
                use tokio::io::{AsyncBufReadExt, BufReader};
                let reader = BufReader::new(&mut stderr);
                let mut lines = reader.lines();

                while let Ok(Some(line)) = lines.next_line().await {
                    // Parser la progression
                    if let Some(progress) = FfmpegProgress::parse(&line) {
                        if let Some(time) = progress.time {
                            tracing::debug!(
                                "Worker {} track {}: time={:?}, speed={:?}x",
                                worker_id,
                                track_id,
                                time,
                                progress.speed
                            );
                        }
                    }

                    // Détecter les erreurs
                    if line.contains("error") || line.contains("Error") || line.contains("ERROR") {
                        tracing::warn!("Worker {} track {} FFmpeg error: {}", worker_id, track_id, line);
                    }
                }
            });
        }

        // 5. Attendre la fin avec timeout
        let result = timeout(JOB_TIMEOUT, child.wait()).await;

        match result {
            Ok(Ok(status)) => {
                if status.success() {
                    // Vérifier que le manifest existe
                    if !manifest_path.exists() {
                        return Err(AppError::InternalError {
                            message: "HLS manifest not generated".to_string(),
                        });
                    }

                    // Parser le manifest et insérer les segments en DB
                    self.parse_and_store_segments(&job).await?;

                    // Mettre à jour le statut
                    self.update_job_status(&job.track_id, EncodeJobStatus::Done).await?;

                    tracing::info!(
                        "Worker {} a terminé l'encodage pour track {} (qualité: {})",
                        self.id,
                        job.track_id,
                        job.quality
                    );
                    Ok(())
                } else {
                    let error_msg = format!("FFmpeg exited with status: {}", status);
                    self.update_job_status_with_error(&job.track_id, &error_msg).await?;
                    Err(AppError::InternalError { message: error_msg })
                }
            }
            Ok(Err(e)) => {
                let error_msg = format!("FFmpeg IO error: {}", e);
                self.update_job_status_with_error(&job.track_id, &error_msg).await?;
                Err(AppError::InternalError { message: error_msg })
            }
            Err(_) => {
                // Timeout: tuer le processus
                let _ = child.kill().await;
                let error_msg = format!("Encoding timed out after {} seconds", JOB_TIMEOUT.as_secs());
                self.update_job_status_with_error(&job.track_id, &error_msg).await?;
                Err(AppError::InternalError { message: error_msg })
            }
        }
    }

    /// Parse le manifest HLS et insère les segments en DB (transactionnel P1)
    /// 
    /// Garantit l'atomicité : tous les segments ou aucun
    /// En cas d'erreur, rollback automatique → pas de playlist HLS incomplète
    async fn parse_and_store_segments(&self, job: &EncodeJob) -> Result<(), AppError> {
        use std::fs;
        use regex::Regex;
        use lazy_static::lazy_static;

        lazy_static! {
            // Regex pour parser les lignes EXTINF et les URIs de segments
            static ref EXTINF_REGEX: Regex = Regex::new(r"#EXTINF:([\d.]+),").unwrap();
        }

        let manifest_path = job.output_dir.join("index.m3u8");
        let manifest_content = fs::read_to_string(&manifest_path)
            .map_err(|e| AppError::InternalError {
                message: format!("Failed to read manifest: {}", e),
            })?;

        // Parser le manifest ligne par ligne
        let lines: Vec<&str> = manifest_content.lines().collect();
        let mut segments_to_insert: Vec<(i32, String, f64)> = Vec::new();
        let mut segment_index = 0;
        let mut current_duration = 4.0; // Durée par défaut (4 secondes)

        // 1. Parser le manifest et collecter les segments
        for line in lines {
            // Parser EXTINF pour obtenir la durée
            if let Some(caps) = EXTINF_REGEX.captures(line) {
                if let Ok(dur) = caps.get(1).unwrap().as_str().parse::<f64>() {
                    current_duration = dur;
                }
            }
            // Ligne qui n'est pas un tag = URI du segment
            else if !line.starts_with('#') && !line.trim().is_empty() {
                let segment_uri = line.trim();
                let segment_path = job.output_dir.join(segment_uri);
                
                if segment_path.exists() {
                    segments_to_insert.push((
                        segment_index,
                        segment_path.to_string_lossy().to_string(),
                        current_duration,
                    ));
                    segment_index += 1;
                }
            }
        }

        if segments_to_insert.is_empty() {
            tracing::warn!(
                "Worker {}: No segments found in manifest for track {} (qualité: {})",
                self.id,
                job.track_id,
                job.quality
            );
            return Ok(());
        }

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

        // 2. VALIDATION : Vérifier que le job existe
        let job_exists: bool = sqlx::query_scalar(
            r#"
            SELECT EXISTS(SELECT 1 FROM stream_jobs WHERE track_id = $1 ORDER BY created_at DESC LIMIT 1)
            "#,
        )
        .bind(job.track_id)
        .fetch_one(&mut *tx)
        .await
        .map_err(|e| AppError::InternalError {
            message: format!("Failed to validate job for track {}: {}", job.track_id, e),
        })?;

        if !job_exists {
            return Err(AppError::NotFound {
                resource: format!("Job for track {}", job.track_id),
            });
        }

        // 3. INSERT tous les segments en batch dans la transaction
        for (index, path, duration) in segments_to_insert.iter() {
            sqlx::query(
                r#"
                INSERT INTO stream_segments (track_id, quality, segment_index, path, duration)
                VALUES ($1, $2, $3, $4, $5)
                ON CONFLICT (track_id, quality, segment_index) DO NOTHING
                "#,
            )
            .bind(job.track_id)
            .bind(&job.quality)
            .bind(index)
            .bind(path)
            .bind(duration)
            .execute(&mut *tx)
            .await
            .map_err(|e| AppError::InternalError {
                message: format!("Failed to insert segment {} in batch: {}", path, e),
            })?;
        }

        // 4. CALCUL : Durée totale
        let total_duration: f64 = segments_to_insert.iter().map(|(_, _, d)| d).sum();

        // 5. UPDATE job (updated_at) - optionnel mais recommandé
        sqlx::query(
            r#"
            UPDATE stream_jobs
            SET updated_at = NOW()
            WHERE id = (
                SELECT id FROM stream_jobs 
                WHERE track_id = $1 
                ORDER BY created_at DESC 
                LIMIT 1
            )
            "#,
        )
        .bind(job.track_id)
        .execute(&mut *tx)
        .await
        .map_err(|e| AppError::InternalError {
            message: format!("Failed to update job after segment insertion: {}", e),
        })?;

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

        tracing::info!(
            worker_id = self.id,
            track_id = %job.track_id,
            quality = %job.quality,
            segment_count = segments_to_insert.len(),
            total_duration = total_duration,
            "All segments inserted successfully in transaction"
        );

        Ok(())
    }

    /// Met à jour le statut d'un job en DB
    async fn update_job_status(
        &self,
        track_id: &Uuid,
        status: EncodeJobStatus,
    ) -> Result<(), AppError> {
        sqlx::query(
            r#"
            UPDATE stream_jobs
            SET status = $1, updated_at = NOW()
            WHERE id = (
                SELECT id FROM stream_jobs 
                WHERE track_id = $2
                ORDER BY created_at DESC 
                LIMIT 1
            )
            "#,
        )
        .bind(status.as_str())
        .bind(track_id)
        .execute(&self.db_pool)
        .await
        .map_err(|e| AppError::InternalError {
            message: format!("Failed to update job status: {}", e),
        })?;

        Ok(())
    }

    /// Met à jour le statut avec un message d'erreur
    async fn update_job_status_with_error(
        &self,
        track_id: &Uuid,
        error_message: &str,
    ) -> Result<(), AppError> {
        sqlx::query(
            r#"
            UPDATE stream_jobs
            SET status = 'error', error_message = $1, updated_at = NOW()
            WHERE id = (
                SELECT id FROM stream_jobs 
                WHERE track_id = $2
                ORDER BY created_at DESC 
                LIMIT 1
            )
            "#,
        )
        .bind(error_message)
        .bind(track_id)
        .execute(&self.db_pool)
        .await
        .map_err(|e| AppError::InternalError {
            message: format!("Failed to update job status with error: {}", e),
        })?;

        Ok(())
    }
}
