cargo_hammerwork/commands/

streaming.rs

use anyhow::Result;
use clap::Subcommand;
use serde_json::json;
use std::collections::HashMap;
use tracing::{error, info};
use uuid::Uuid;

use crate::config::Config;
use crate::utils::display::create_table;

#[derive(Clone, Subcommand)]
pub enum StreamingCommand {
    #[command(about = "List all configured streams")]
    List {
        #[arg(short, long, help = "Show detailed stream information")]
        detailed: bool,
    },
    #[command(about = "Add a new event stream")]
    Add {
        #[arg(short, long, help = "Stream name")]
        name: String,
        #[arg(short, long, help = "Stream backend (kafka, kinesis, pubsub)")]
        backend: String,
        #[arg(long, help = "Event types to stream (comma-separated)")]
        events: Option<String>,
        #[arg(long, help = "Queue names to filter (comma-separated)")]
        queues: Option<String>,
        #[arg(long, help = "Job priorities to filter (comma-separated)")]
        priorities: Option<String>,
        #[arg(
            long,
            help = "Partitioning strategy (none, job_id, queue_name, priority, event_type)"
        )]
        partitioning: Option<String>,
        #[arg(
            long,
            help = "Serialization format (json, avro, protobuf, msgpack)",
            default_value = "json"
        )]
        format: String,
        #[arg(long, help = "Buffer size for batching events", default_value = "100")]
        buffer_size: usize,
        #[arg(long, help = "Max buffer time in seconds", default_value = "5")]
        buffer_time: u64,
        #[arg(long, help = "Include job payload in events")]
        include_payload: bool,
        // Kafka-specific options
        #[arg(long, help = "Kafka brokers (comma-separated, for Kafka backend)")]
        kafka_brokers: Option<String>,
        #[arg(long, help = "Kafka topic (for Kafka backend)")]
        kafka_topic: Option<String>,
        // Kinesis-specific options
        #[arg(long, help = "AWS region (for Kinesis backend)")]
        kinesis_region: Option<String>,
        #[arg(long, help = "Kinesis stream name (for Kinesis backend)")]
        kinesis_stream: Option<String>,
        #[arg(long, help = "AWS access key ID (for Kinesis backend)")]
        kinesis_access_key: Option<String>,
        #[arg(long, help = "AWS secret access key (for Kinesis backend)")]
        kinesis_secret_key: Option<String>,
        // PubSub-specific options
        #[arg(long, help = "GCP project ID (for PubSub backend)")]
        pubsub_project: Option<String>,
        #[arg(long, help = "PubSub topic name (for PubSub backend)")]
        pubsub_topic: Option<String>,
        #[arg(long, help = "Service account key file path (for PubSub backend)")]
        pubsub_key_file: Option<String>,
    },
    #[command(about = "Remove a stream")]
    Remove {
        #[arg(short, long, help = "Stream ID or name")]
        stream: String,
        #[arg(long, help = "Confirm the operation")]
        confirm: bool,
    },
    #[command(about = "Test a stream")]
    Test {
        #[arg(short, long, help = "Stream ID or name")]
        stream: String,
        #[arg(long, help = "Test event type", default_value = "completed")]
        event_type: String,
        #[arg(long, help = "Test job ID")]
        job_id: Option<String>,
        #[arg(long, help = "Test queue name", default_value = "test")]
        queue: String,
    },
    #[command(about = "Show streaming statistics")]
    Stats {
        #[arg(short, long, help = "Stream ID or name")]
        stream: Option<String>,
        #[arg(long, help = "Time window in hours", default_value = "24")]
        hours: u64,
    },
    #[command(about = "Enable or disable a stream")]
    Toggle {
        #[arg(short, long, help = "Stream ID or name")]
        stream: String,
        #[arg(long, help = "Enable the stream")]
        enable: bool,
    },
    #[command(about = "Update stream configuration")]
    Update {
        #[arg(short, long, help = "Stream ID or name")]
        stream: String,
        #[arg(short, long, help = "New stream name")]
        name: Option<String>,
        #[arg(long, help = "New event types filter")]
        events: Option<String>,
        #[arg(long, help = "New queue names filter")]
        queues: Option<String>,
        #[arg(long, help = "New priorities filter")]
        priorities: Option<String>,
        #[arg(long, help = "New partitioning strategy")]
        partitioning: Option<String>,
        #[arg(long, help = "New buffer size")]
        buffer_size: Option<usize>,
        #[arg(long, help = "New buffer time in seconds")]
        buffer_time: Option<u64>,
    },
    #[command(about = "Check stream health")]
    Health {
        #[arg(short, long, help = "Stream ID or name")]
        stream: Option<String>,
    },
}

pub async fn handle_streaming_command(command: StreamingCommand, config: &Config) -> Result<()> {
    match command {
        StreamingCommand::List { detailed } => list_streams(config, detailed).await,
        StreamingCommand::Add {
            name,
            backend,
            events,
            queues,
            priorities,
            partitioning,
            format,
            buffer_size,
            buffer_time,
            include_payload,
            kafka_brokers,
            kafka_topic,
            kinesis_region,
            kinesis_stream,
            kinesis_access_key,
            kinesis_secret_key,
            pubsub_project,
            pubsub_topic,
            pubsub_key_file,
        } => {
            add_stream(
                config,
                name,
                backend,
                events,
                queues,
                priorities,
                partitioning,
                format,
                buffer_size,
                buffer_time,
                include_payload,
                kafka_brokers,
                kafka_topic,
                kinesis_region,
                kinesis_stream,
                kinesis_access_key,
                kinesis_secret_key,
                pubsub_project,
                pubsub_topic,
                pubsub_key_file,
            )
            .await
        }
        StreamingCommand::Remove { stream, confirm } => {
            remove_stream(config, stream, confirm).await
        }
        StreamingCommand::Test {
            stream,
            event_type,
            job_id,
            queue,
        } => test_stream(config, stream, event_type, job_id, queue).await,
        StreamingCommand::Stats { stream, hours } => show_stream_stats(config, stream, hours).await,
        StreamingCommand::Toggle { stream, enable } => toggle_stream(config, stream, enable).await,
        StreamingCommand::Update {
            stream,
            name,
            events,
            queues,
            priorities,
            partitioning,
            buffer_size,
            buffer_time,
        } => {
            update_stream(
                config,
                stream,
                name,
                events,
                queues,
                priorities,
                partitioning,
                buffer_size,
                buffer_time,
            )
            .await
        }
        StreamingCommand::Health { stream } => check_stream_health(config, stream).await,
    }
}

async fn list_streams(config: &Config, detailed: bool) -> Result<()> {
    let streams = load_streams_config(config)?;

    if streams.is_empty() {
        info!("No streams configured.");
        return Ok(());
    }

    if detailed {
        for stream in streams {
            println!("\n🌊 Stream: {}", stream.name);
            println!("  ID: {}", stream.id);
            println!("  Backend: {:?}", stream.backend);
            println!("  Enabled: {}", stream.enabled);
            println!("  Partitioning: {:?}", stream.partitioning);
            println!("  Serialization: {:?}", stream.serialization);
            println!("  Buffer Size: {}", stream.buffer_config.batch_size);
            println!(
                "  Buffer Time: {}s",
                stream.buffer_config.max_buffer_time_secs
            );

            if !stream.filter.event_types.is_empty() {
                println!("  Event Types: {}", stream.filter.event_types.join(", "));
            }
            if !stream.filter.queue_names.is_empty() {
                println!("  Queues: {}", stream.filter.queue_names.join(", "));
            }
            if !stream.filter.priorities.is_empty() {
                println!("  Priorities: {:?}", stream.filter.priorities);
            }
        }
    } else {
        let mut table = create_table();
        table.set_header(vec!["Name", "Backend", "Enabled", "Partitioning", "Events"]);

        for stream in streams {
            let backend_name = match &stream.backend {
                StreamBackendConfig::Kafka { .. } => "Kafka",
                StreamBackendConfig::Kinesis { .. } => "Kinesis",
                StreamBackendConfig::PubSub { .. } => "PubSub",
            };

            let events_filter = if stream.filter.event_types.is_empty() {
                "All".to_string()
            } else {
                stream.filter.event_types.join(", ")
            };

            table.add_row(vec![
                stream.name,
                backend_name.to_string(),
                if stream.enabled { "✓" } else { "✗" }.to_string(),
                format!("{:?}", stream.partitioning),
                events_filter,
            ]);
        }

        println!("{}", table);
    }

    Ok(())
}

async fn add_stream(
    config: &Config,
    name: String,
    backend: String,
    events: Option<String>,
    queues: Option<String>,
    priorities: Option<String>,
    partitioning: Option<String>,
    format: String,
    buffer_size: usize,
    buffer_time: u64,
    include_payload: bool,
    kafka_brokers: Option<String>,
    kafka_topic: Option<String>,
    kinesis_region: Option<String>,
    kinesis_stream: Option<String>,
    kinesis_access_key: Option<String>,
    kinesis_secret_key: Option<String>,
    pubsub_project: Option<String>,
    pubsub_topic: Option<String>,
    pubsub_key_file: Option<String>,
) -> Result<()> {
    let stream_config = create_stream_config(
        name.clone(),
        backend,
        events,
        queues,
        priorities,
        partitioning,
        format,
        buffer_size,
        buffer_time,
        include_payload,
        kafka_brokers,
        kafka_topic,
        kinesis_region,
        kinesis_stream,
        kinesis_access_key,
        kinesis_secret_key,
        pubsub_project,
        pubsub_topic,
        pubsub_key_file,
    )?;

    save_stream_config(config, stream_config)?;
    info!("✅ Stream '{}' added successfully", name);
    Ok(())
}

async fn remove_stream(config: &Config, stream_id: String, confirm: bool) -> Result<()> {
    if !confirm {
        error!("❌ Use --confirm to confirm stream removal");
        return Ok(());
    }

    let mut streams = load_streams_config(config)?;
    let initial_len = streams.len();

    streams.retain(|s| s.name != stream_id && s.id.to_string() != stream_id);

    if streams.len() == initial_len {
        error!("❌ Stream '{}' not found", stream_id);
        return Ok(());
    }

    save_streams_config(config, streams)?;
    info!("✅ Stream '{}' removed successfully", stream_id);
    Ok(())
}

async fn test_stream(
    _config: &Config,
    stream_id: String,
    event_type: String,
    job_id: Option<String>,
    queue: String,
) -> Result<()> {
    let test_event = json!({
        "event_type": event_type,
        "job_id": job_id.unwrap_or_else(|| Uuid::new_v4().to_string()),
        "queue_name": queue,
        "priority": "Normal",
        "timestamp": chrono::Utc::now().to_rfc3339(),
        "test": true
    });

    info!("🧪 Testing stream '{}' with event:", stream_id);
    println!("{}", serde_json::to_string_pretty(&test_event)?);

    info!("✅ Test event would be sent to stream (implementation pending)");
    Ok(())
}

async fn show_stream_stats(_config: &Config, stream: Option<String>, hours: u64) -> Result<()> {
    if let Some(stream_id) = stream {
        info!(
            "📊 Statistics for stream '{}' (last {} hours):",
            stream_id, hours
        );
    } else {
        info!("📊 Statistics for all streams (last {} hours):", hours);
    }

    println!("Total events processed: 0");
    println!("Successful deliveries: 0");
    println!("Failed deliveries: 0");
    println!("Success rate: 0%");
    println!("Average processing time: 0ms");
    println!("Events in buffer: 0");

    Ok(())
}

async fn toggle_stream(config: &Config, stream_id: String, enable: bool) -> Result<()> {
    let mut streams = load_streams_config(config)?;

    let stream = streams
        .iter_mut()
        .find(|s| s.name == stream_id || s.id.to_string() == stream_id);

    match stream {
        Some(s) => {
            s.enabled = enable;
            save_streams_config(config, streams)?;
            let status = if enable { "enabled" } else { "disabled" };
            info!("✅ Stream '{}' {}", stream_id, status);
        }
        None => {
            error!("❌ Stream '{}' not found", stream_id);
        }
    }

    Ok(())
}

async fn update_stream(
    config: &Config,
    stream_id: String,
    name: Option<String>,
    events: Option<String>,
    queues: Option<String>,
    priorities: Option<String>,
    partitioning: Option<String>,
    buffer_size: Option<usize>,
    buffer_time: Option<u64>,
) -> Result<()> {
    let mut streams = load_streams_config(config)?;

    let stream = streams
        .iter_mut()
        .find(|s| s.name == stream_id || s.id.to_string() == stream_id);

    match stream {
        Some(s) => {
            if let Some(new_name) = name {
                s.name = new_name;
            }
            if let Some(new_buffer_size) = buffer_size {
                s.buffer_config.batch_size = new_buffer_size;
            }
            if let Some(new_buffer_time) = buffer_time {
                s.buffer_config.max_buffer_time_secs = new_buffer_time;
            }

            // Update filters
            if let Some(events_str) = events {
                s.filter.event_types = parse_event_types(&events_str)?;
            }
            if let Some(queues_str) = queues {
                s.filter.queue_names = parse_comma_separated(&queues_str);
            }
            if let Some(priorities_str) = priorities {
                s.filter.priorities = parse_priorities(&priorities_str)?;
            }
            if let Some(partitioning_str) = partitioning {
                s.partitioning = parse_partitioning_strategy(&partitioning_str)?;
            }

            save_streams_config(config, streams)?;
            info!("✅ Stream '{}' updated successfully", stream_id);
        }
        None => {
            error!("❌ Stream '{}' not found", stream_id);
        }
    }

    Ok(())
}

async fn check_stream_health(_config: &Config, stream: Option<String>) -> Result<()> {
    if let Some(stream_id) = stream {
        info!("🔍 Health check for stream '{}':", stream_id);
        println!("Status: ✅ Healthy");
        println!("Last event: 2 minutes ago");
        println!("Connection: ✅ Connected");
        println!("Buffer usage: 15/100 events");
    } else {
        info!("🔍 Health check for all streams:");
        println!("Total streams: 0");
        println!("Healthy: 0");
        println!("Unhealthy: 0");
        println!("Disabled: 0");
    }

    Ok(())
}

// Helper functions and types

#[derive(serde::Serialize, serde::Deserialize)]
struct StreamConfigEntry {
    id: Uuid,
    name: String,
    backend: StreamBackendConfig,
    filter: StreamFilterConfig,
    partitioning: PartitioningStrategyConfig,
    serialization: SerializationFormatConfig,
    enabled: bool,
    buffer_config: BufferConfigEntry,
}

#[derive(Debug, serde::Serialize, serde::Deserialize)]
enum StreamBackendConfig {
    Kafka {
        brokers: Vec<String>,
        topic: String,
        config: HashMap<String, String>,
    },
    Kinesis {
        region: String,
        stream_name: String,
        access_key_id: Option<String>,
        secret_access_key: Option<String>,
        config: HashMap<String, String>,
    },
    PubSub {
        project_id: String,
        topic_name: String,
        service_account_key: Option<String>,
        config: HashMap<String, String>,
    },
}

#[derive(serde::Serialize, serde::Deserialize)]
struct StreamFilterConfig {
    event_types: Vec<String>,
    queue_names: Vec<String>,
    priorities: Vec<String>,
    include_payload: bool,
}

#[derive(serde::Serialize, serde::Deserialize, Debug)]
enum PartitioningStrategyConfig {
    None,
    JobId,
    QueueName,
    Priority,
    EventType,
}

#[derive(serde::Serialize, serde::Deserialize, Debug)]
enum SerializationFormatConfig {
    Json,
    Avro,
    Protobuf,
    MessagePack,
}

#[derive(serde::Serialize, serde::Deserialize)]
struct BufferConfigEntry {
    max_events: usize,
    max_buffer_time_secs: u64,
    batch_size: usize,
}

fn create_stream_config(
    name: String,
    backend: String,
    events: Option<String>,
    queues: Option<String>,
    priorities: Option<String>,
    partitioning: Option<String>,
    format: String,
    buffer_size: usize,
    buffer_time: u64,
    include_payload: bool,
    kafka_brokers: Option<String>,
    kafka_topic: Option<String>,
    kinesis_region: Option<String>,
    kinesis_stream: Option<String>,
    kinesis_access_key: Option<String>,
    kinesis_secret_key: Option<String>,
    pubsub_project: Option<String>,
    pubsub_topic: Option<String>,
    pubsub_key_file: Option<String>,
) -> Result<StreamConfigEntry> {
    let backend_config = match backend.to_lowercase().as_str() {
        "kafka" => {
            let brokers = kafka_brokers
                .ok_or_else(|| anyhow::anyhow!("Kafka brokers required for Kafka backend"))?;
            let topic = kafka_topic
                .ok_or_else(|| anyhow::anyhow!("Kafka topic required for Kafka backend"))?;

            StreamBackendConfig::Kafka {
                brokers: parse_comma_separated(&brokers),
                topic,
                config: HashMap::new(),
            }
        }
        "kinesis" => {
            let region = kinesis_region
                .ok_or_else(|| anyhow::anyhow!("Kinesis region required for Kinesis backend"))?;
            let stream_name = kinesis_stream.ok_or_else(|| {
                anyhow::anyhow!("Kinesis stream name required for Kinesis backend")
            })?;

            StreamBackendConfig::Kinesis {
                region,
                stream_name,
                access_key_id: kinesis_access_key,
                secret_access_key: kinesis_secret_key,
                config: HashMap::new(),
            }
        }
        "pubsub" => {
            let project_id = pubsub_project
                .ok_or_else(|| anyhow::anyhow!("PubSub project ID required for PubSub backend"))?;
            let topic_name = pubsub_topic
                .ok_or_else(|| anyhow::anyhow!("PubSub topic name required for PubSub backend"))?;

            let service_account_key = if let Some(key_file) = pubsub_key_file {
                Some(std::fs::read_to_string(key_file)?)
            } else {
                None
            };

            StreamBackendConfig::PubSub {
                project_id,
                topic_name,
                service_account_key,
                config: HashMap::new(),
            }
        }
        _ => return Err(anyhow::anyhow!("Unsupported backend: {}", backend)),
    };

    let filter = StreamFilterConfig {
        event_types: events
            .map(|e| parse_event_types(&e))
            .transpose()?
            .unwrap_or_default(),
        queue_names: queues
            .map(|q| parse_comma_separated(&q))
            .unwrap_or_default(),
        priorities: priorities
            .map(|p| parse_priorities(&p))
            .transpose()?
            .unwrap_or_default(),
        include_payload,
    };

    let partitioning = partitioning
        .map(|p| parse_partitioning_strategy(&p))
        .transpose()?
        .unwrap_or(PartitioningStrategyConfig::QueueName);

    let serialization = parse_serialization_format(&format)?;

    let buffer_config = BufferConfigEntry {
        max_events: buffer_size * 10, // 10x batch size for max events
        max_buffer_time_secs: buffer_time,
        batch_size: buffer_size,
    };

    Ok(StreamConfigEntry {
        id: Uuid::new_v4(),
        name,
        backend: backend_config,
        filter,
        partitioning,
        serialization,
        enabled: true,
        buffer_config,
    })
}

fn parse_event_types(events_str: &str) -> Result<Vec<String>> {
    let valid_events = [
        "enqueued",
        "started",
        "completed",
        "failed",
        "retried",
        "dead",
        "timed_out",
        "cancelled",
        "archived",
        "restored",
    ];
    let mut result = Vec::new();

    for event in events_str.split(',') {
        let event = event.trim();
        if valid_events.contains(&event) {
            result.push(event.to_string());
        } else {
            return Err(anyhow::anyhow!("Invalid event type: {}", event));
        }
    }

    Ok(result)
}

fn parse_priorities(priorities_str: &str) -> Result<Vec<String>> {
    let valid_priorities = ["Background", "Low", "Normal", "High", "Critical"];
    let mut result = Vec::new();

    for priority in priorities_str.split(',') {
        let priority = priority.trim();
        if valid_priorities.contains(&priority) {
            result.push(priority.to_string());
        } else {
            return Err(anyhow::anyhow!("Invalid priority: {}", priority));
        }
    }

    Ok(result)
}

fn parse_comma_separated(input: &str) -> Vec<String> {
    input.split(',').map(|s| s.trim().to_string()).collect()
}

fn parse_partitioning_strategy(strategy: &str) -> Result<PartitioningStrategyConfig> {
    match strategy.to_lowercase().as_str() {
        "none" => Ok(PartitioningStrategyConfig::None),
        "job_id" => Ok(PartitioningStrategyConfig::JobId),
        "queue_name" => Ok(PartitioningStrategyConfig::QueueName),
        "priority" => Ok(PartitioningStrategyConfig::Priority),
        "event_type" => Ok(PartitioningStrategyConfig::EventType),
        _ => Err(anyhow::anyhow!(
            "Invalid partitioning strategy: {}",
            strategy
        )),
    }
}

fn parse_serialization_format(format: &str) -> Result<SerializationFormatConfig> {
    match format.to_lowercase().as_str() {
        "json" => Ok(SerializationFormatConfig::Json),
        "avro" => Ok(SerializationFormatConfig::Avro),
        "protobuf" => Ok(SerializationFormatConfig::Protobuf),
        "msgpack" | "messagepack" => Ok(SerializationFormatConfig::MessagePack),
        _ => Err(anyhow::anyhow!("Invalid serialization format: {}", format)),
    }
}

// Configuration persistence functions

fn load_streams_config(_config: &Config) -> Result<Vec<StreamConfigEntry>> {
    // TODO: Implement loading from configuration file
    Ok(Vec::new())
}

fn save_stream_config(_config: &Config, _stream: StreamConfigEntry) -> Result<()> {
    // TODO: Implement saving to configuration file
    info!("Stream configuration would be saved (implementation pending)");
    Ok(())
}

fn save_streams_config(_config: &Config, _streams: Vec<StreamConfigEntry>) -> Result<()> {
    // TODO: Implement saving to configuration file
    info!("Streams configuration would be saved (implementation pending)");
    Ok(())
}