//! Text domain module for NexusCompress.
//!
//! Provides `TextCompressor` for compressing batches of text messages
//! using length-prefixed encoding and zstd compression.

use nexuscompress_codecs::create_codec;
use nexuscompress_core::plan::{FieldPipeline, PlanCodecNode};
use nexuscompress_core::{CodecContext, CompressError, DType, Domain, Frame, Plan, Schema};

/// Configuration for text compression.
pub struct TextConfig {
    /// Maximum dictionary size hint (in bytes) for the compressor.
    pub max_dict_size: usize,
}

impl Default for TextConfig {
    fn default() -> Self {
        Self {
            max_dict_size: 65536,
        }
    }
}

/// Compressor for batches of text messages.
pub struct TextCompressor {
    config: TextConfig,
}

impl TextCompressor {
    /// Create a new `TextCompressor` with the given configuration.
    pub fn new(config: TextConfig) -> Self {
        Self { config }
    }

    /// Encode messages into a single byte buffer with u32 LE length prefixes.
    fn encode_messages(messages: &[&str]) -> Vec<u8> {
        let total: usize = messages.iter().map(|m| 4 + m.len()).sum();
        let mut buf = Vec::with_capacity(total);
        for msg in messages {
            buf.extend_from_slice(&(msg.len() as u32).to_le_bytes());
            buf.extend_from_slice(msg.as_bytes());
        }
        buf
    }

    /// Decode length-prefixed messages from a byte buffer.
    fn decode_messages(data: &[u8]) -> Result<Vec<String>, CompressError> {
        let mut messages = Vec::new();
        let mut pos = 0;
        while pos + 4 <= data.len() {
            let len = u32::from_le_bytes(data[pos..pos + 4].try_into().map_err(|_| {
                CompressError::InvalidFrame {
                    message: "truncated message length".into(),
                }
            })?) as usize;
            pos += 4;
            if pos + len > data.len() {
                return Err(CompressError::InvalidFrame {
                    message: "truncated message data".into(),
                });
            }
            let s = String::from_utf8(data[pos..pos + len].to_vec()).map_err(|e| {
                CompressError::CodecError {
                    codec: "text".into(),
                    message: format!("invalid UTF-8: {e}"),
                }
            })?;
            messages.push(s);
            pos += len;
        }
        Ok(messages)
    }

    /// Compress a batch of text messages.
    ///
    /// Returns serialized `Frame` bytes.
    pub fn compress_batch(&self, messages: &[&str]) -> Result<Vec<u8>, CompressError> {
        if messages.is_empty() {
            return Err(CompressError::EmptyInput);
        }

        // Length-prefix encode all messages
        let encoded = Self::encode_messages(messages);

        // Compress with zstd
        let zstd_codec = create_codec("zstd", &[])?;
        let ctx = CodecContext::new(DType::U8, encoded.len())
            .with_metadata("max_dict_size", self.config.max_dict_size.to_string());
        let mut compressed = Vec::new();
        zstd_codec.compress(&encoded, &mut compressed, &ctx)?;

        // Build plan
        let schema = Schema::builder()
            .type_name("TextBatch")
            .domain(Domain::Text)
            .field("messages", DType::Bytes)
            .build();

        let mut plan = Plan::new("text", Domain::Text, schema);
        plan.graph.add_pipeline(FieldPipeline {
            field_name: "messages".into(),
            field_index: 0,
            codecs: vec![PlanCodecNode {
                name: "zstd_stage".into(),
                codec_name: "zstd".into(),
                config: Vec::new(),
            }],
        });

        let mut frame = Frame::new(&plan, messages.len() as u64)?;
        frame.add_field_data(compressed);
        frame.serialize()
    }

    /// Decompress a frame back into individual text messages.
    pub fn decompress_batch(&self, frame_bytes: &[u8]) -> Result<Vec<String>, CompressError> {
        let frame = Frame::deserialize(frame_bytes)?;
        let plan = frame.embedded_plan()?;

        if frame.field_data.is_empty() {
            return Ok(Vec::new());
        }

        let compressed = &frame.field_data[0];

        // Find the pipeline and apply codecs in reverse
        let pipeline = plan
            .graph
            .pipelines
            .iter()
            .find(|p| p.field_index == 0)
            .ok_or_else(|| CompressError::InvalidFrame {
                message: "no pipeline for messages field".into(),
            })?;

        let ctx = CodecContext::new(DType::U8, compressed.len());
        let mut current = compressed.clone();
        for codec_node in pipeline.codecs.iter().rev() {
            let codec = create_codec(&codec_node.codec_name, &codec_node.config)?;
            let mut out = Vec::new();
            codec.decompress(&current, &mut out, &ctx)?;
            current = out;
        }

        Self::decode_messages(&current)
    }
}

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

    #[test]
    fn round_trip_text_messages() {
        let compressor = TextCompressor::new(TextConfig::default());

        let messages = vec![
            "Hello, world!",
            "This is a test message.",
            "The quick brown fox jumps over the lazy dog.",
            "NexusCompress provides structured data compression.",
            "Rust is a systems programming language.",
        ];

        let compressed = compressor.compress_batch(&messages).unwrap();
        assert!(!compressed.is_empty());

        let decompressed = compressor.decompress_batch(&compressed).unwrap();
        assert_eq!(decompressed.len(), messages.len());
        for (orig, dec) in messages.iter().zip(decompressed.iter()) {
            assert_eq!(*orig, dec.as_str());
        }
    }

    #[test]
    fn empty_input_error() {
        let compressor = TextCompressor::new(TextConfig::default());
        let result = compressor.compress_batch(&[]);
        assert!(result.is_err());
    }

    #[test]
    fn single_message() {
        let compressor = TextCompressor::new(TextConfig::default());

        let messages = vec!["only one message"];
        let compressed = compressor.compress_batch(&messages).unwrap();
        let decompressed = compressor.decompress_batch(&compressed).unwrap();
        assert_eq!(decompressed, vec!["only one message"]);
    }

    #[test]
    fn empty_string_message() {
        let compressor = TextCompressor::new(TextConfig::default());

        let messages = vec!["", "non-empty", ""];
        let compressed = compressor.compress_batch(&messages).unwrap();
        let decompressed = compressor.decompress_batch(&compressed).unwrap();
        assert_eq!(decompressed, vec!["", "non-empty", ""]);
    }

    #[test]
    fn large_batch() {
        let compressor = TextCompressor::new(TextConfig::default());

        let messages_owned: Vec<String> = (0..1000)
            .map(|i| {
                format!(
                    "Message number {} with some repeated content for compression",
                    i
                )
            })
            .collect();
        let messages: Vec<&str> = messages_owned.iter().map(|s| s.as_str()).collect();

        let compressed = compressor.compress_batch(&messages).unwrap();

        // Similar messages should compress well
        let raw_size: usize = messages.iter().map(|m| m.len()).sum();
        assert!(compressed.len() < raw_size);

        let decompressed = compressor.decompress_batch(&compressed).unwrap();
        assert_eq!(decompressed.len(), 1000);
        for (orig, dec) in messages.iter().zip(decompressed.iter()) {
            assert_eq!(*orig, dec.as_str());
        }
    }

    #[test]
    fn unicode_messages() {
        let compressor = TextCompressor::new(TextConfig::default());

        let messages = vec!["Bonjour le monde!", "Hallo Welt!", "Hola Mundo!"];

        let compressed = compressor.compress_batch(&messages).unwrap();
        let decompressed = compressor.decompress_batch(&compressed).unwrap();
        assert_eq!(decompressed.len(), 3);
        for (orig, dec) in messages.iter().zip(decompressed.iter()) {
            assert_eq!(*orig, dec.as_str());
        }
    }

    #[test]
    fn default_config_values() {
        let config = TextConfig::default();
        assert_eq!(config.max_dict_size, 65536);
    }

    #[test]
    fn custom_config() {
        let config = TextConfig {
            max_dict_size: 1024,
        };
        let compressor = TextCompressor::new(config);

        let messages = vec!["test with small dict"];
        let compressed = compressor.compress_batch(&messages).unwrap();
        let decompressed = compressor.decompress_batch(&compressed).unwrap();
        assert_eq!(decompressed, vec!["test with small dict"]);
    }

    #[test]
    fn encode_decode_round_trip() {
        let messages = vec!["one", "two", "three"];
        let encoded = TextCompressor::encode_messages(&messages);
        let decoded = TextCompressor::decode_messages(&encoded).unwrap();
        assert_eq!(decoded, vec!["one", "two", "three"]);
    }

    #[test]
    fn decode_messages_truncated_length() {
        // Less than 4 bytes: should return empty.
        let result = TextCompressor::decode_messages(&[0u8, 1]);
        assert!(result.unwrap().is_empty());
    }

    #[test]
    fn decode_messages_truncated_data() {
        let mut data = Vec::new();
        data.extend_from_slice(&100u32.to_le_bytes());
        data.extend_from_slice(b"short");
        let result = TextCompressor::decode_messages(&data);
        assert!(result.is_err());
    }

    #[test]
    fn round_trip_with_newlines() {
        let compressor = TextCompressor::new(TextConfig::default());
        let messages = vec!["line1\nline2\nline3", "another\nmultiline\nmessage"];
        let compressed = compressor.compress_batch(&messages).unwrap();
        let decompressed = compressor.decompress_batch(&compressed).unwrap();
        assert_eq!(decompressed.len(), 2);
        assert_eq!(decompressed[0], messages[0]);
        assert_eq!(decompressed[1], messages[1]);
    }

    #[test]
    fn round_trip_cjk_text() {
        let compressor = TextCompressor::new(TextConfig::default());
        let messages = vec![
            "\u{4f60}\u{597d}\u{4e16}\u{754c}",
            "\u{3053}\u{3093}\u{306b}\u{3061}\u{306f}",
        ];
        let compressed = compressor.compress_batch(&messages).unwrap();
        let decompressed = compressor.decompress_batch(&compressed).unwrap();
        assert_eq!(decompressed[0], messages[0]);
        assert_eq!(decompressed[1], messages[1]);
    }

    #[test]
    fn round_trip_many_empty_strings() {
        let compressor = TextCompressor::new(TextConfig::default());
        let messages = vec![""; 50];
        let compressed = compressor.compress_batch(&messages).unwrap();
        let decompressed = compressor.decompress_batch(&compressed).unwrap();
        assert_eq!(decompressed.len(), 50);
        for msg in &decompressed {
            assert!(msg.is_empty());
        }
    }

    #[test]
    fn encode_empty_messages_round_trip() {
        let messages: Vec<&str> = vec!["", "", ""];
        let encoded = TextCompressor::encode_messages(&messages);
        let decoded = TextCompressor::decode_messages(&encoded).unwrap();
        assert_eq!(decoded, vec!["", "", ""]);
    }
}
