use sqlx::{PgPool, Row};
use serde::{Deserialize, Serialize};
use std::collections::HashMap;
use std::sync::Arc;
use tokio::sync::RwLock;
use tokio::time::{Duration, Instant};
use tracing::{info, warn, error, debug};
use uuid::Uuid;

#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct Message {
    pub id: Uuid,
    pub conversation_id: Uuid,
    pub user_id: Uuid,
    pub content: String,
    pub message_type: String, // "text", "audio", "image", etc.
    pub metadata: Option<serde_json::Value>,
    pub created_at: chrono::DateTime<chrono::Utc>,
    pub updated_at: chrono::DateTime<chrono::Utc>,
}

#[derive(Debug, Clone)]
pub struct PendingMessage {
    pub conversation_id: Uuid,
    pub user_id: Uuid,
    pub content: String,
    pub message_type: String,
    pub metadata: Option<serde_json::Value>,
    pub created_at: chrono::DateTime<chrono::Utc>,
}

#[derive(Debug)]
pub struct MessageStore {
    pool: PgPool,
    batch_buffer: Arc<RwLock<Vec<PendingMessage>>>,
    batch_size: usize,
    flush_interval: Duration,
    last_flush: Arc<RwLock<Instant>>,
}

impl MessageStore {
    pub fn new(pool: PgPool) -> Self {
        Self {
            pool,
            batch_buffer: Arc::new(RwLock::new(Vec::new())),
            batch_size: 50, // Flush après 50 messages
            flush_interval: Duration::from_millis(100), // Ou après 100ms
            last_flush: Arc::new(RwLock::new(Instant::now())),
        }
    }

    /// Ajoute un message au buffer et flush si nécessaire
    pub async fn add_message(&self, message: PendingMessage) -> Result<(), sqlx::Error> {
        let mut buffer = self.batch_buffer.write().await;
        buffer.push(message);

        // Vérifier si on doit flush
        let should_flush = buffer.len() >= self.batch_size;
        let time_since_last_flush = self.last_flush.read().await.elapsed();

        if should_flush || time_since_last_flush >= self.flush_interval {
            drop(buffer); // Libérer le lock avant le flush
            self.flush_buffer().await?;
        }

        Ok(())
    }

    /// Flush le buffer vers la base de données
    async fn flush_buffer(&self) -> Result<(), sqlx::Error> {
        let mut buffer = self.batch_buffer.write().await;
        if buffer.is_empty() {
            return Ok(());
        }

        let messages = buffer.clone();
        buffer.clear();
        drop(buffer);

        // Mettre à jour le timestamp du dernier flush
        *self.last_flush.write().await = Instant::now();

        // Utiliser COPY pour un insert bulk optimisé
        self.bulk_insert_messages(messages).await
    }

    /// Insert bulk optimisé avec COPY
    async fn bulk_insert_messages(&self, messages: Vec<PendingMessage>) -> Result<(), sqlx::Error> {
        if messages.is_empty() {
            return Ok(());
        }

        debug!("Bulk inserting {} messages", messages.len());

        // Utiliser une transaction pour l'atomicité
        let mut tx = self.pool.begin().await?;

        // Construire la requête COPY
        let mut copy_data = String::new();
        for msg in &messages {
            copy_data.push_str(&format!(
                "{}\t{}\t{}\t{}\t{}\t{}\t{}\t{}\n",
                Uuid::new_v4(), // id
                msg.conversation_id,
                msg.user_id,
                msg.content.replace('\t', "\\t").replace('\n', "\\n"),
                msg.message_type,
                msg.metadata.as_ref().map(|m| m.to_string()).unwrap_or("\\N".to_string()),
                msg.created_at.format("%Y-%m-%d %H:%M:%S%.f %z"),
                msg.created_at.format("%Y-%m-%d %H:%M:%S%.f %z"), // updated_at
            ));
        }

        // Exécuter le COPY
        sqlx::query(&format!(
            "COPY messages (id, conversation_id, user_id, content, message_type, metadata, created_at, updated_at) FROM STDIN WITH (FORMAT text)"
        ))
        .execute(&mut *tx)
        .await?;

        tx.commit().await?;

        info!("Successfully bulk inserted {} messages", messages.len());
        Ok(())
    }

    /// Récupère les messages d'une conversation avec pagination
    pub async fn get_messages(
        &self,
        conversation_id: Uuid,
        limit: i64,
        offset: i64,
    ) -> Result<Vec<Message>, sqlx::Error> {
        let rows = sqlx::query(
            r#"
            SELECT id, conversation_id, user_id, content, message_type, metadata, created_at, updated_at
            FROM messages
            WHERE conversation_id = $1
            ORDER BY created_at DESC
            LIMIT $2 OFFSET $3
            "#,
        )
        .bind(conversation_id)
        .bind(limit)
        .bind(offset)
        .fetch_all(&self.pool)
        .await?;

        let mut messages = Vec::new();
        for row in rows {
            let message = Message {
                id: row.get("id"),
                conversation_id: row.get("conversation_id"),
                user_id: row.get("user_id"),
                content: row.get("content"),
                message_type: row.get("message_type"),
                metadata: row.get("metadata"),
                created_at: row.get("created_at"),
                updated_at: row.get("updated_at"),
            };
            messages.push(message);
        }

        Ok(messages)
    }

    /// Récupère les messages récents d'une conversation (pour le cache)
    pub async fn get_recent_messages(
        &self,
        conversation_id: Uuid,
        limit: i64,
    ) -> Result<Vec<Message>, sqlx::Error> {
        self.get_messages(conversation_id, limit, 0).await
    }

    /// Supprime les messages anciens (nettoyage)
    pub async fn cleanup_old_messages(&self, older_than_days: i32) -> Result<u64, sqlx::Error> {
        let result = sqlx::query(
            "DELETE FROM messages WHERE created_at < NOW() - INTERVAL '%s days'",
        )
        .bind(older_than_days)
        .execute(&self.pool)
        .await?;

        let deleted_count = result.rows_affected();
        info!("Cleaned up {} old messages", deleted_count);
        Ok(deleted_count)
    }

    /// Force le flush du buffer (utile pour les tests ou l'arrêt)
    pub async fn force_flush(&self) -> Result<(), sqlx::Error> {
        self.flush_buffer().await
    }

    /// Obtient les statistiques du buffer
    pub async fn get_buffer_stats(&self) -> (usize, Duration) {
        let buffer_len = self.batch_buffer.read().await.len();
        let time_since_last_flush = self.last_flush.read().await.elapsed();
        (buffer_len, time_since_last_flush)
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use chrono::Utc;

    #[tokio::test]
    async fn test_message_store_batch_insert() {
        // Test avec une base de données en mémoire ou mock
        // Vérifier que les messages sont bien batchés
        let pool = PgPool::connect("postgresql://test:test@localhost/test").await.unwrap();
        let store = MessageStore::new(pool);

        let message = PendingMessage {
            conversation_id: Uuid::new_v4(),
            user_id: Uuid::new_v4(),
            content: "Test message".to_string(),
            message_type: "text".to_string(),
            metadata: None,
            created_at: Utc::now(),
        };

        // Ajouter plusieurs messages
        for _ in 0..60 {
            store.add_message(message.clone()).await.unwrap();
        }

        // Vérifier que le buffer a été flushé
        let (buffer_len, _) = store.get_buffer_stats().await;
        assert!(buffer_len < 50); // Buffer devrait être flushé
    }
}