fluxmq 0.1.0

use bytes::Bytes;
use fluxmq::{
    broker::MessageHandler,
    protocol::{FetchRequest, Message, ProduceRequest, Request, Response},
};
use std::sync::Arc;

#[tokio::test]
async fn test_large_message_handling() {
    let handler = MessageHandler::new().unwrap();

    // Create a very large message (1MB)
    let large_content = "x".repeat(1024 * 1024);
    let large_message = Message::new(large_content.clone());

    let produce_req = ProduceRequest {
        correlation_id: 1,
        topic: "large-message-topic".to_string(),
        partition: 0,
        messages: vec![large_message],
        acks: 1,
        timeout_ms: 10000, // Longer timeout for large message
    };

    // Should handle large message without error
    let response = handler
        .handle_request(Request::Produce(produce_req))
        .await
        .expect("Failed to handle large message");

    match response {
        Response::Produce(resp) => {
            assert_eq!(resp.error_code, 0);
            assert_eq!(resp.base_offset, 0);
        }
        _ => panic!("Expected produce response"),
    }

    // Fetch the large message back
    let fetch_req = FetchRequest {
        correlation_id: 2,
        topic: "large-message-topic".to_string(),
        partition: 0,
        offset: 0,
        max_bytes: 2 * 1024 * 1024, // 2MB to accommodate the large message
        timeout_ms: 10000,
    };

    let response = handler
        .handle_request(Request::Fetch(fetch_req))
        .await
        .expect("Failed to fetch large message");

    match response {
        Response::Fetch(fetch_resp) => {
            assert_eq!(fetch_resp.error_code, 0);
            assert_eq!(fetch_resp.messages.len(), 1);
            assert_eq!(fetch_resp.messages[0].1.value, Bytes::from(large_content));
        }
        _ => panic!("Expected fetch response"),
    }
}

#[tokio::test]
async fn test_max_bytes_limit_in_fetch() {
    let handler = MessageHandler::new().unwrap();

    // Produce multiple messages
    let mut messages = Vec::new();
    for i in 0..10 {
        messages.push(Message::new(format!("message_{}", i).repeat(100))); // ~1KB each
    }

    let produce_req = ProduceRequest {
        correlation_id: 1,
        topic: "limited-fetch-topic".to_string(),
        partition: 0,
        messages,
        acks: 1,
        timeout_ms: 5000,
    };

    handler
        .handle_request(Request::Produce(produce_req))
        .await
        .expect("Failed to produce messages");

    // Fetch with very small max_bytes to test limiting
    let fetch_req = FetchRequest {
        correlation_id: 2,
        topic: "limited-fetch-topic".to_string(),
        partition: 0,
        offset: 0,
        max_bytes: 500, // Should limit the number of messages returned
        timeout_ms: 3000,
    };

    let response = handler
        .handle_request(Request::Fetch(fetch_req))
        .await
        .expect("Failed to fetch with limit");

    match response {
        Response::Fetch(fetch_resp) => {
            assert_eq!(fetch_resp.error_code, 0);
            // Should return fewer messages due to size limit (or potentially empty if none fit)
            assert!(fetch_resp.messages.len() <= 10);
            // The current implementation may return empty if no messages fit the byte limit
            // This is acceptable behavior for a simple implementation
        }
        _ => panic!("Expected fetch response"),
    }
}

#[tokio::test]
async fn test_empty_message_batch() {
    let handler = MessageHandler::new().unwrap();

    // Try to produce empty message batch
    let produce_req = ProduceRequest {
        correlation_id: 1,
        topic: "empty-batch-topic".to_string(),
        partition: 0,
        messages: vec![], // Empty batch
        acks: 1,
        timeout_ms: 5000,
    };

    let response = handler
        .handle_request(Request::Produce(produce_req))
        .await
        .expect("Should handle empty batch gracefully");

    match response {
        Response::Produce(resp) => {
            assert_eq!(resp.error_code, 0);
            // Base offset should still be valid even for empty batch
            assert_eq!(resp.base_offset, 0);
        }
        _ => panic!("Expected produce response"),
    }
}

#[tokio::test]
async fn test_fetch_beyond_available_messages() {
    let handler = MessageHandler::new().unwrap();

    // Produce only 3 messages
    let messages = vec![
        Message::new("msg1"),
        Message::new("msg2"),
        Message::new("msg3"),
    ];

    let produce_req = ProduceRequest {
        correlation_id: 1,
        topic: "limited-topic".to_string(),
        partition: 0,
        messages,
        acks: 1,
        timeout_ms: 5000,
    };

    handler
        .handle_request(Request::Produce(produce_req))
        .await
        .expect("Failed to produce messages");

    // Try to fetch from offset beyond available messages
    let fetch_req = FetchRequest {
        correlation_id: 2,
        topic: "limited-topic".to_string(),
        partition: 0,
        offset: 10, // Beyond available messages
        max_bytes: 1024,
        timeout_ms: 3000,
    };

    let response = handler
        .handle_request(Request::Fetch(fetch_req))
        .await
        .expect("Should handle out-of-range fetch gracefully");

    match response {
        Response::Fetch(fetch_resp) => {
            assert_eq!(fetch_resp.error_code, 0);
            assert_eq!(fetch_resp.messages.len(), 0); // Should return empty
        }
        _ => panic!("Expected fetch response"),
    }
}

#[tokio::test]
async fn test_invalid_correlation_ids() {
    let handler = MessageHandler::new().unwrap();

    // Test with negative correlation ID
    let message = Message::new("test message");
    let produce_req = ProduceRequest {
        correlation_id: -999,
        topic: "correlation-test".to_string(),
        partition: 0,
        messages: vec![message],
        acks: 1,
        timeout_ms: 5000,
    };

    let response = handler
        .handle_request(Request::Produce(produce_req))
        .await
        .expect("Should handle negative correlation ID");

    match response {
        Response::Produce(resp) => {
            assert_eq!(resp.correlation_id, -999); // Should preserve the ID
            assert_eq!(resp.error_code, 0);
        }
        _ => panic!("Expected produce response"),
    }
}

#[tokio::test]
async fn test_very_long_topic_names() {
    let handler = MessageHandler::new().unwrap();

    // Test with very long topic name (but reasonable)
    let long_topic = "a".repeat(1000);
    let message = Message::new("test message");

    let produce_req = ProduceRequest {
        correlation_id: 1,
        topic: long_topic.clone(),
        partition: 0,
        messages: vec![message],
        acks: 1,
        timeout_ms: 5000,
    };

    let response = handler
        .handle_request(Request::Produce(produce_req))
        .await
        .expect("Should handle long topic names");

    match response {
        Response::Produce(resp) => {
            assert_eq!(resp.topic, long_topic);
            assert_eq!(resp.error_code, 0);
        }
        _ => panic!("Expected produce response"),
    }
}

#[tokio::test]
async fn test_high_partition_numbers() {
    let handler = MessageHandler::new().unwrap();

    // Test with very high partition number
    let message = Message::new("partition test");
    let produce_req = ProduceRequest {
        correlation_id: 1,
        topic: "high-partition-topic".to_string(),
        partition: u32::MAX - 1, // Very high partition number
        messages: vec![message],
        acks: 1,
        timeout_ms: 5000,
    };

    let response = handler
        .handle_request(Request::Produce(produce_req))
        .await
        .expect("Should handle high partition numbers");

    match response {
        Response::Produce(resp) => {
            assert_eq!(resp.partition, u32::MAX - 1);
            assert_eq!(resp.error_code, 3); // Should return unknown partition error
        }
        _ => panic!("Expected produce response"),
    }
}

#[tokio::test]
async fn test_zero_timeout() {
    let handler = MessageHandler::new().unwrap();

    let message = Message::new("timeout test");
    let produce_req = ProduceRequest {
        correlation_id: 1,
        topic: "timeout-topic".to_string(),
        partition: 0,
        messages: vec![message],
        acks: 1,
        timeout_ms: 0, // Zero timeout
    };

    // Should still process the request even with zero timeout
    let response = handler
        .handle_request(Request::Produce(produce_req))
        .await
        .expect("Should handle zero timeout");

    match response {
        Response::Produce(resp) => {
            assert_eq!(resp.error_code, 0);
        }
        _ => panic!("Expected produce response"),
    }
}

#[tokio::test]
async fn test_message_with_empty_value() {
    let handler = MessageHandler::new().unwrap();

    // Create message with empty value
    let empty_message = Message::new("");
    let produce_req = ProduceRequest {
        correlation_id: 1,
        topic: "empty-value-topic".to_string(),
        partition: 0,
        messages: vec![empty_message],
        acks: 1,
        timeout_ms: 5000,
    };

    let response = handler
        .handle_request(Request::Produce(produce_req))
        .await
        .expect("Should handle empty message values");

    match response {
        Response::Produce(resp) => {
            assert_eq!(resp.error_code, 0);
        }
        _ => panic!("Expected produce response"),
    }

    // Fetch it back
    let fetch_req = FetchRequest {
        correlation_id: 2,
        topic: "empty-value-topic".to_string(),
        partition: 0,
        offset: 0,
        max_bytes: 1024,
        timeout_ms: 3000,
    };

    let response = handler
        .handle_request(Request::Fetch(fetch_req))
        .await
        .expect("Should fetch empty message");

    match response {
        Response::Fetch(fetch_resp) => {
            assert_eq!(fetch_resp.messages.len(), 1);
            assert_eq!(fetch_resp.messages[0].1.value, Bytes::from(""));
        }
        _ => panic!("Expected fetch response"),
    }
}

#[tokio::test]
async fn test_recovery_after_operations() {
    let handler = Arc::new(MessageHandler::new().unwrap());

    // Perform a series of operations
    for i in 0..10 {
        let message = Message::new(format!("recovery_test_{}", i));
        let produce_req = ProduceRequest {
            correlation_id: i,
            topic: "recovery-topic".to_string(),
            partition: 0,
            messages: vec![message],
            acks: 1,
            timeout_ms: 5000,
        };

        handler
            .handle_request(Request::Produce(produce_req))
            .await
            .expect("Failed during operation sequence");
    }

    // Verify system is still functional after many operations
    let fetch_req = FetchRequest {
        correlation_id: 100,
        topic: "recovery-topic".to_string(),
        partition: 0,
        offset: 0,
        max_bytes: 10240,
        timeout_ms: 3000,
    };

    let response = handler
        .handle_request(Request::Fetch(fetch_req))
        .await
        .expect("System should still be functional");

    match response {
        Response::Fetch(fetch_resp) => {
            assert_eq!(fetch_resp.error_code, 0);
            assert_eq!(fetch_resp.messages.len(), 10);

            // Verify message integrity
            for (i, (offset, message)) in fetch_resp.messages.iter().enumerate() {
                assert_eq!(*offset, i as u64);
                assert_eq!(message.value, Bytes::from(format!("recovery_test_{}", i)));
            }
        }
        _ => panic!("Expected fetch response"),
    }
}