hammerwork 2.3.0

//! Event streaming integration for external message systems.
//!
//! This module provides streaming integrations for delivering job lifecycle events
//! to external message systems like Apache Kafka, AWS Kinesis, and Google Cloud Pub/Sub.
//! It supports configurable routing, partitioning, and delivery guarantees.
//!
//! # Overview
//!
//! The streaming system consists of several key components:
//!
//! - [`StreamManager`] - Manages stream configurations and delivery to external systems
//! - [`StreamConfig`] - Configuration for individual stream endpoints
//! - [`StreamProcessor`] - Backend-specific processors (Kafka, Kinesis, PubSub)
//! - [`PartitioningStrategy`] - Configurable event partitioning for distributed systems
//! - [`SerializationFormat`] - Multiple serialization options (JSON, Avro, Protobuf)
//! - [`BufferConfig`] - Configurable buffering and batching for performance
//!
//! # Features
//!
//! - **Multiple streaming backends** - Kafka, AWS Kinesis, Google Cloud Pub/Sub
//! - **Flexible partitioning** - Partition by job ID, queue name, priority, event type, or custom fields
//! - **Multiple serialization formats** - JSON, Avro with schema registry, Protocol Buffers, MessagePack
//! - **Configurable buffering** - Buffer events for batch delivery with size and time limits
//! - **Retry policies** - Exponential backoff with jitter for failed deliveries
//! - **Delivery tracking** - Comprehensive statistics and delivery history
//! - **Rate limiting** - Configurable concurrent processing limits
//! - **Health monitoring** - Backend health checks and failover support
//!
//! # Examples
//!
//! ## Basic Kafka Streaming Setup
//!
//! ```rust
//! use hammerwork::streaming::{StreamManager, StreamConfig, StreamBackend, PartitioningStrategy, SerializationFormat, StreamManagerConfig};
//! use hammerwork::events::{EventManager, EventFilter, JobLifecycleEventType};
//! use std::sync::Arc;
//! use std::collections::HashMap;
//!
//! # #[tokio::main]
//! # async fn main() -> Result<(), Box<dyn std::error::Error>> {
//! let event_manager = Arc::new(EventManager::new_default());
//! let stream_manager = StreamManager::new(
//!     event_manager.clone(),
//!     StreamManagerConfig::default()
//! );
//!
//! let stream = StreamConfig {
//!     id: uuid::Uuid::new_v4(),
//!     name: "job_events_kafka".to_string(),
//!     backend: StreamBackend::Kafka {
//!         brokers: vec!["localhost:9092".to_string()],
//!         topic: "hammerwork-job-events".to_string(),
//!         config: HashMap::new(),
//!     },
//!     filter: EventFilter::new()
//!         .with_event_types(vec![JobLifecycleEventType::Completed, JobLifecycleEventType::Failed]),
//!     partitioning: PartitioningStrategy::QueueName,
//!     serialization: SerializationFormat::Json,
//!     enabled: true,
//!     ..Default::default()
//! };
//!
//! stream_manager.add_stream(stream).await?;
//! # Ok(())
//! # }
//! ```
//!
//! ## AWS Kinesis with Custom Partitioning
//!
//! ```rust
//! use hammerwork::streaming::{StreamConfig, StreamBackend, PartitioningStrategy, PartitionField, SerializationFormat};
//! use hammerwork::events::EventFilter;
//! use std::collections::HashMap;
//!
//! let stream = StreamConfig {
//!     id: uuid::Uuid::new_v4(),
//!     name: "priority_events_kinesis".to_string(),
//!     backend: StreamBackend::Kinesis {
//!         region: "us-east-1".to_string(),
//!         stream_name: "hammerwork-priority-events".to_string(),
//!         access_key_id: None, // Use IAM roles
//!         secret_access_key: None,
//!         config: HashMap::new(),
//!     },
//!     filter: EventFilter::new()
//!         .with_priorities(vec![hammerwork::priority::JobPriority::High, hammerwork::priority::JobPriority::Critical]),
//!     partitioning: PartitioningStrategy::Hash {
//!         fields: vec![PartitionField::Priority, PartitionField::QueueName]
//!     },
//!     serialization: SerializationFormat::Json,
//!     enabled: true,
//!     ..Default::default()
//! };
//! ```
//!
//! ## Google Cloud Pub/Sub with Buffering
//!
//! ```rust
//! use hammerwork::streaming::{StreamConfig, StreamBackend, BufferConfig, SerializationFormat};
//! use hammerwork::events::EventFilter;
//! use std::collections::HashMap;
//!
//! let stream = StreamConfig {
//!     id: uuid::Uuid::new_v4(),
//!     name: "analytics_events_pubsub".to_string(),
//!     backend: StreamBackend::PubSub {
//!         project_id: "my-project".to_string(),
//!         topic_name: "hammerwork-analytics".to_string(),
//!         service_account_key: None, // Use default credentials
//!         config: HashMap::new(),
//!     },
//!     filter: EventFilter::new().include_payload(),
//!     buffer_config: BufferConfig {
//!         max_events: 500,
//!         max_buffer_time_secs: 10,
//!         batch_size: 50,
//!     },
//!     serialization: SerializationFormat::Json,
//!     enabled: true,
//!     ..Default::default()
//! };
//! ```
//!
//! ## Avro Serialization with Schema Registry
//!
//! ```rust
//! use hammerwork::streaming::{StreamConfig, SerializationFormat};
//!
//! let stream = StreamConfig {
//!     serialization: SerializationFormat::Avro {
//!         schema_registry_url: "http://localhost:8081".to_string()
//!     },
//!     ..Default::default()
//! };
//! ```

use crate::{
    HammerworkError,
    events::{EventFilter, EventManager, JobLifecycleEvent},
    task_tracker::{ListenerHandle, TaskTracker, clamp_permits},
};

#[cfg(feature = "kafka")]
use rdkafka::{
    ClientConfig,
    message::{Header, OwnedHeaders},
    producer::{FutureProducer, FutureRecord, Producer},
};

#[cfg(feature = "google-pubsub")]
use google_cloud_pubsub::{
    client::{Client, ClientConfig as PubSubClientConfig},
    publisher::Publisher,
    topic::Topic,
};

#[cfg(feature = "google-pubsub")]
use google_cloud_pubsub::client::google_cloud_auth::credentials::CredentialsFile;

#[cfg(feature = "google-pubsub")]
use google_cloud_googleapis::pubsub::v1::PubsubMessage;

#[cfg(feature = "kinesis")]
use aws_sdk_kinesis::{Client as KinesisClient, config::Region};

#[cfg(feature = "kinesis")]
use aws_config::BehaviorVersion;
use chrono::{DateTime, Utc};
use serde::{Deserialize, Serialize};
use std::{
    collections::{HashMap, HashSet},
    sync::Arc,
    time::Duration,
};
use tokio::sync::{OwnedSemaphorePermit, RwLock, Semaphore, broadcast, oneshot, watch};
use uuid::Uuid;

/// Processors by stream id. Processors are reference-counted so a batch can send
/// through one without holding the map lock across network I/O.
type ProcessorMap = Arc<RwLock<HashMap<Uuid, Arc<dyn StreamProcessor + Send + Sync>>>>;

/// Module for serializing UUID as string for TOML compatibility
mod uuid_string {
    use serde::{Deserialize, Deserializer, Serializer};
    use uuid::Uuid;

    pub fn serialize<S>(uuid: &Uuid, serializer: S) -> Result<S::Ok, S::Error>
    where
        S: Serializer,
    {
        serializer.serialize_str(&uuid.to_string())
    }

    pub fn deserialize<'de, D>(deserializer: D) -> Result<Uuid, D::Error>
    where
        D: Deserializer<'de>,
    {
        use serde::de::Error;
        let s = String::deserialize(deserializer)?;
        Uuid::parse_str(&s).map_err(D::Error::custom)
    }
}

/// Configuration for event streaming.
///
/// StreamConfig defines how events should be delivered to a specific streaming backend.
/// It includes backend configuration, event filtering, partitioning strategy, serialization
/// format, and delivery options.
///
/// # Examples
///
/// ```rust
/// use hammerwork::streaming::{StreamConfig, StreamBackend, PartitioningStrategy, SerializationFormat, BufferConfig};
/// use hammerwork::events::EventFilter;
/// use uuid::Uuid;
/// use std::collections::HashMap;
///
/// // Basic Kafka configuration
/// let kafka_stream = StreamConfig {
///     id: Uuid::new_v4(),
///     name: "main_events".to_string(),
///     backend: StreamBackend::Kafka {
///         brokers: vec!["localhost:9092".to_string()],
///         topic: "job-events".to_string(),
///         config: HashMap::new(),
///     },
///     filter: EventFilter::new(),
///     partitioning: PartitioningStrategy::QueueName,
///     serialization: SerializationFormat::Json,
///     enabled: true,
///     ..Default::default()
/// };
///
/// // Kinesis with custom partitioning
/// let kinesis_stream = StreamConfig {
///     id: Uuid::new_v4(),
///     name: "analytics_stream".to_string(),
///     backend: StreamBackend::Kinesis {
///         region: "us-west-2".to_string(),
///         stream_name: "analytics-events".to_string(),
///         access_key_id: None,
///         secret_access_key: None,
///         config: HashMap::new(),
///     },
///     partitioning: PartitioningStrategy::JobId,
///     buffer_config: BufferConfig {
///         max_events: 100,
///         max_buffer_time_secs: 5,
///         batch_size: 25,
///     },
///     ..Default::default()
/// };
/// ```
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct StreamConfig {
    /// Unique identifier for this stream
    #[serde(with = "uuid_string")]
    pub id: Uuid,
    /// Human-readable name for this stream
    pub name: String,
    /// Streaming backend configuration
    pub backend: StreamBackend,
    /// Event filter to determine which events to stream
    pub filter: EventFilter,
    /// Partitioning strategy for the stream
    pub partitioning: PartitioningStrategy,
    /// Serialization format for events
    pub serialization: SerializationFormat,
    /// Retry policy for failed deliveries
    pub retry_policy: StreamRetryPolicy,
    /// Whether this stream is currently enabled
    pub enabled: bool,
    /// Buffer configuration
    pub buffer_config: BufferConfig,
}

/// Streaming backend types.
///
/// StreamBackend defines the configuration for different streaming systems.
/// Each backend has its own specific configuration options and capabilities.
///
/// # Supported Backends
///
/// - **Kafka** - Apache Kafka with configurable brokers and topics
/// - **Kinesis** - AWS Kinesis Data Streams with IAM or credential-based authentication
/// - **PubSub** - Google Cloud Pub/Sub with service account or default credentials
///
/// # Examples
///
/// ```rust
/// use hammerwork::streaming::StreamBackend;
/// use std::collections::HashMap;
///
/// // Kafka backend
/// let kafka = StreamBackend::Kafka {
///     brokers: vec!["broker1:9092".to_string(), "broker2:9092".to_string()],
///     topic: "events".to_string(),
///     config: {
///         let mut config = HashMap::new();
///         config.insert("acks".to_string(), "all".to_string());
///         config.insert("retries".to_string(), "3".to_string());
///         config
///     },
/// };
///
/// // Kinesis with IAM roles
/// let kinesis = StreamBackend::Kinesis {
///     region: "us-east-1".to_string(),
///     stream_name: "my-event-stream".to_string(),
///     access_key_id: None, // Use IAM roles
///     secret_access_key: None,
///     config: HashMap::new(),
/// };
///
/// // Pub/Sub with default credentials
/// let pubsub = StreamBackend::PubSub {
///     project_id: "my-gcp-project".to_string(),
///     topic_name: "event-topic".to_string(),
///     service_account_key: None, // Use default credentials
///     config: HashMap::new(),
/// };
/// ```
///
/// `Debug` never shows credentials (`secret_access_key`, `service_account_key`) or the
/// values of the `config` maps (they may hold SASL passwords), only their keys.
#[derive(Clone, Serialize, Deserialize)]
#[serde(tag = "type")]
pub enum StreamBackend {
    /// Apache Kafka backend
    Kafka {
        /// Kafka broker addresses
        brokers: Vec<String>,
        /// Topic to publish events to
        topic: String,
        /// Additional Kafka configuration
        config: HashMap<String, String>,
    },
    /// AWS Kinesis backend
    Kinesis {
        /// AWS region
        region: String,
        /// Kinesis stream name
        stream_name: String,
        /// AWS access key ID (optional, can use IAM roles)
        access_key_id: Option<String>,
        /// AWS secret access key (optional, can use IAM roles)
        secret_access_key: Option<String>,
        /// Additional Kinesis configuration
        config: HashMap<String, String>,
    },
    /// Google Cloud Pub/Sub backend
    PubSub {
        /// GCP project ID
        project_id: String,
        /// Pub/Sub topic name
        topic_name: String,
        /// Service account key JSON (optional, can use default credentials)
        service_account_key: Option<String>,
        /// Additional Pub/Sub configuration
        config: HashMap<String, String>,
    },
}

impl std::fmt::Debug for StreamBackend {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        fn keys(map: &HashMap<String, String>) -> Vec<&String> {
            let mut keys: Vec<&String> = map.keys().collect();
            keys.sort();
            keys
        }
        let redacted = |secret: &Option<String>| secret.as_ref().map(|_| "[REDACTED]");
        match self {
            StreamBackend::Kafka {
                brokers,
                topic,
                config,
            } => f
                .debug_struct("Kafka")
                .field("brokers", brokers)
                .field("topic", topic)
                .field("config", &keys(config))
                .finish(),
            StreamBackend::Kinesis {
                region,
                stream_name,
                access_key_id,
                secret_access_key,
                config,
            } => f
                .debug_struct("Kinesis")
                .field("region", region)
                .field("stream_name", stream_name)
                .field("access_key_id", access_key_id)
                .field("secret_access_key", &redacted(secret_access_key))
                .field("config", &keys(config))
                .finish(),
            StreamBackend::PubSub {
                project_id,
                topic_name,
                service_account_key,
                config,
            } => f
                .debug_struct("PubSub")
                .field("project_id", project_id)
                .field("topic_name", topic_name)
                .field("service_account_key", &redacted(service_account_key))
                .field("config", &keys(config))
                .finish(),
        }
    }
}

/// Partitioning strategies for distributing events across stream partitions.
///
/// Partitioning strategies determine how events are distributed across partitions
/// in the streaming backend. This affects message ordering, load distribution,
/// and processing patterns.
///
/// # Strategies
///
/// - **None** - All events go to a single partition (preserves total ordering)
/// - **JobId** - Partition by job ID (events for same job stay together)
/// - **QueueName** - Partition by queue name (events from same queue stay together)
/// - **Priority** - Partition by job priority (similar priorities grouped)
/// - **EventType** - Partition by event type (completions, failures, etc. grouped)
/// - **Custom** - Partition by custom metadata field
/// - **Hash** - Hash-based partitioning using multiple fields
///
/// # Examples
///
/// ```rust
/// use hammerwork::streaming::{PartitioningStrategy, PartitionField};
///
/// // No partitioning - all events in order
/// let no_partition = PartitioningStrategy::None;
///
/// // Partition by queue name
/// let queue_partition = PartitioningStrategy::QueueName;
///
/// // Custom metadata partitioning
/// let custom_partition = PartitioningStrategy::Custom {
///     metadata_key: "tenant_id".to_string()
/// };
///
/// // Hash-based partitioning with multiple fields
/// let hash_partition = PartitioningStrategy::Hash {
///     fields: vec![
///         PartitionField::QueueName,
///         PartitionField::Priority,
///         PartitionField::MetadataKey("customer_id".to_string())
///     ]
/// };
/// ```
#[derive(Debug, Clone, Serialize, Deserialize)]
#[serde(tag = "type")]
pub enum PartitioningStrategy {
    /// No partitioning - all events go to single partition
    None,
    /// Partition by job ID
    JobId,
    /// Partition by queue name
    QueueName,
    /// Partition by job priority
    Priority,
    /// Partition by event type
    EventType,
    /// Custom partitioning key from event metadata
    Custom { metadata_key: String },
    /// Hash-based partitioning using multiple fields
    Hash { fields: Vec<PartitionField> },
}

/// Fields that can be used for hash-based partitioning
#[derive(Debug, Clone, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum PartitionField {
    JobId,
    QueueName,
    Priority,
    EventType,
    MetadataKey(String),
}

/// Serialization formats for event data.
///
/// SerializationFormat determines how job lifecycle events are serialized
/// before being sent to the streaming backend. Different formats have
/// different trade-offs in terms of size, performance, and compatibility.
///
/// # Formats
///
/// - **JSON** - Human-readable, widely compatible, larger size
/// - **Avro** - Schema evolution support, compact binary format
/// - **Protobuf** - Efficient binary format, strong typing
/// - **MessagePack** - Compact binary JSON-like format
///
/// # Examples
///
/// ```rust
/// use hammerwork::streaming::SerializationFormat;
///
/// // JSON serialization (default)
/// let json_format = SerializationFormat::Json;
///
/// // Avro with schema registry
/// let avro_format = SerializationFormat::Avro {
///     schema_registry_url: "http://localhost:8081".to_string()
/// };
///
/// // Protocol Buffers with embedded schema
/// let protobuf_format = SerializationFormat::Protobuf {
///     schema_definition: r#"
///         syntax = "proto3";
///         message JobEvent {
///             string job_id = 1;
///             string event_type = 2;
///             // ... other fields
///         }
///     "#.to_string()
/// };
///
/// // MessagePack for compact binary
/// let msgpack_format = SerializationFormat::MessagePack;
/// ```
#[derive(Debug, Clone, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum SerializationFormat {
    /// JSON serialization
    Json,
    /// Avro serialization (with schema registry)
    Avro { schema_registry_url: String },
    /// Protocol Buffers serialization
    Protobuf { schema_definition: String },
    /// MessagePack serialization
    MessagePack,
}

/// Retry policy for streaming failures.
///
/// StreamRetryPolicy defines how failed streaming deliveries should be retried.
/// It supports exponential backoff with optional jitter to prevent thundering herd problems.
///
/// # Examples
///
/// ```rust
/// use hammerwork::streaming::StreamRetryPolicy;
///
/// // Default retry policy
/// let default_policy = StreamRetryPolicy::default();
///
/// // Custom retry policy for critical events
/// let aggressive_policy = StreamRetryPolicy {
///     max_attempts: 10,
///     initial_delay_secs: 1,
///     max_delay_secs: 600, // 10 minutes
///     backoff_multiplier: 2.0,
///     use_jitter: true,
/// };
///
/// // Quick retry policy for non-critical events
/// let fast_policy = StreamRetryPolicy {
///     max_attempts: 3,
///     initial_delay_secs: 1,
///     max_delay_secs: 30,
///     backoff_multiplier: 1.5,
///     use_jitter: false,
/// };
/// ```
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct StreamRetryPolicy {
    /// Maximum number of delivery attempts per event, including the first one.
    /// `0` is treated as `1` (a single attempt, no retries).
    pub max_attempts: u32,
    /// Initial delay between retries (in seconds)
    pub initial_delay_secs: u64,
    /// Maximum delay between retries (in seconds)
    pub max_delay_secs: u64,
    /// Multiplier for exponential backoff
    pub backoff_multiplier: f64,
    /// Whether to use jitter to avoid thundering herd
    pub use_jitter: bool,
}

impl Default for StreamRetryPolicy {
    fn default() -> Self {
        Self {
            max_attempts: 5,
            initial_delay_secs: 1,
            max_delay_secs: 300, // 5 minutes
            backoff_multiplier: 2.0,
            use_jitter: true,
        }
    }
}

impl StreamRetryPolicy {
    /// Delay before retry number `retry` (1 for the first retry).
    ///
    /// The delay is `initial_delay_secs * backoff_multiplier^(retry - 1)`, capped at
    /// `max_delay_secs`. With `use_jitter` the delay is drawn uniformly from
    /// `[delay / 2, delay]` so that streams failing together do not retry in lockstep.
    pub fn delay_for_retry(&self, retry: u32) -> Duration {
        let exponent = i32::try_from(retry.saturating_sub(1)).unwrap_or(i32::MAX);
        let multiplier = if self.backoff_multiplier.is_finite() && self.backoff_multiplier > 0.0 {
            self.backoff_multiplier
        } else {
            1.0
        };
        let max = self.max_delay_secs as f64;
        let mut delay = (self.initial_delay_secs as f64 * multiplier.powi(exponent)).min(max);
        if !delay.is_finite() || delay < 0.0 {
            delay = max;
        }
        if self.use_jitter && delay > 0.0 {
            use rand::Rng;
            delay = rand::thread_rng().gen_range(delay / 2.0..=delay);
        }
        Duration::try_from_secs_f64(delay).unwrap_or(Duration::from_secs(self.max_delay_secs))
    }
}

/// Buffer configuration for streaming.
///
/// BufferConfig controls how events are buffered and batched before being sent
/// to the streaming backend. Proper buffering can significantly improve throughput
/// while managing memory usage and delivery latency.
///
/// # Examples
///
/// ```rust
/// use hammerwork::streaming::BufferConfig;
///
/// // Default buffer configuration
/// let default_config = BufferConfig::default();
///
/// // High-throughput configuration
/// let high_throughput = BufferConfig {
///     max_events: 5000,
///     max_buffer_time_secs: 10,
///     batch_size: 500,
/// };
///
/// // Low-latency configuration
/// let low_latency = BufferConfig {
///     max_events: 100,
///     max_buffer_time_secs: 1,
///     batch_size: 10,
/// };
///
/// // Memory-constrained configuration
/// let memory_efficient = BufferConfig {
///     max_events: 50,
///     max_buffer_time_secs: 2,
///     batch_size: 5,
/// };
/// ```
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct BufferConfig {
    /// Maximum number of events to buffer
    pub max_events: usize,
    /// Maximum time to buffer events before forcing a flush (in seconds)
    pub max_buffer_time_secs: u64,
    /// Batch size for sending events
    pub batch_size: usize,
}

impl Default for BufferConfig {
    fn default() -> Self {
        Self {
            max_events: 1000,
            max_buffer_time_secs: 5,
            batch_size: 100,
        }
    }
}

impl Default for StreamConfig {
    fn default() -> Self {
        Self {
            id: Uuid::new_v4(),
            name: "default_stream".to_string(),
            backend: StreamBackend::Kafka {
                brokers: vec!["localhost:9092".to_string()],
                topic: "hammerwork-events".to_string(),
                config: HashMap::new(),
            },
            filter: EventFilter::new(),
            partitioning: PartitioningStrategy::None,
            serialization: SerializationFormat::Json,
            retry_policy: StreamRetryPolicy::default(),
            enabled: true,
            buffer_config: BufferConfig::default(),
        }
    }
}

/// A streamed event with routing metadata
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct StreamedEvent {
    /// The original job lifecycle event
    pub event: JobLifecycleEvent,
    /// Partition key for this event
    pub partition_key: Option<String>,
    /// Serialized event data
    pub serialized_data: Vec<u8>,
    /// Timestamp when the event was streamed
    pub streamed_at: DateTime<Utc>,
    /// Headers/metadata for the stream
    pub headers: HashMap<String, String>,
}

/// Stream delivery attempt result
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct StreamDelivery {
    /// Unique identifier for this delivery attempt
    #[serde(with = "uuid_string")]
    pub delivery_id: Uuid,
    /// Stream configuration ID
    #[serde(with = "uuid_string")]
    pub stream_id: Uuid,
    /// Event that was delivered
    #[serde(with = "uuid_string")]
    pub event_id: Uuid,
    /// Whether the delivery was successful
    pub success: bool,
    /// Error message if delivery failed
    pub error_message: Option<String>,
    /// When the delivery was attempted
    pub attempted_at: DateTime<Utc>,
    /// How long the delivery took (in milliseconds)
    pub duration_ms: Option<u64>,
    /// Attempt number (1-based)
    pub attempt_number: u32,
    /// Partition the event was sent to (if applicable)
    pub partition: Option<String>,
}

/// Statistics for a streaming configuration
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct StreamStats {
    /// Stream configuration ID
    #[serde(with = "uuid_string")]
    pub stream_id: Uuid,
    /// Total number of events processed
    pub total_events: u64,
    /// Number of successfully delivered events
    pub successful_deliveries: u64,
    /// Number of failed deliveries
    pub failed_deliveries: u64,
    /// Current success rate (0.0 to 1.0)
    pub success_rate: f64,
    /// Average delivery time in milliseconds
    pub avg_delivery_time_ms: f64,
    /// Events currently in buffer
    pub buffered_events: u64,
    /// Last successful delivery timestamp
    pub last_success_at: Option<DateTime<Utc>>,
    /// Last failure timestamp
    pub last_failure_at: Option<DateTime<Utc>>,
    /// Error from the most recent failed delivery (after retries were exhausted)
    #[serde(default)]
    pub last_error: Option<String>,
    /// Events this stream never processed because its listener fell behind: while
    /// [`BufferConfig::max_events`] events are being delivered the listener waits, and
    /// events that overflow the event manager's buffer in the meantime are dropped
    /// (oldest first).
    #[serde(default)]
    pub dropped_events: u64,
    /// Statistics calculation timestamp
    pub calculated_at: DateTime<Utc>,
}

/// Overall stream manager statistics  
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct StreamManagerGlobalStats {
    /// Total number of configured streams
    pub total_streams: usize,
    /// Number of active/enabled streams
    pub active_streams: usize,
    /// Total events processed across all streams
    pub total_events: u64,
    /// Total successful deliveries across all streams
    pub successful_deliveries: u64,
    /// Total failed deliveries across all streams
    pub failed_deliveries: u64,
}

/// Event streaming manager.
///
/// StreamManager is the central component for managing streaming integrations.
/// It handles stream configurations, event subscriptions, backend processors,
/// delivery tracking, and performance monitoring.
///
/// # Features
///
/// - **Multi-backend support** - Manages Kafka, Kinesis, and Pub/Sub processors
/// - **Concurrent processing** - Rate-limited concurrent event delivery
/// - **Statistics tracking** - Per-stream and global delivery statistics
/// - **Health monitoring** - Backend health checks and failover
/// - **Event filtering** - Only processes events matching stream filters
/// - **Graceful shutdown** - Clean shutdown of all processors and subscriptions
///
/// # Examples
///
/// ```rust
/// use hammerwork::streaming::{StreamManager, StreamManagerConfig};
/// use hammerwork::events::EventManager;
/// use std::sync::Arc;
///
/// # #[tokio::main]
/// # async fn main() -> Result<(), Box<dyn std::error::Error>> {
/// let event_manager = Arc::new(EventManager::new_default());
/// let stream_manager = StreamManager::new(
///     event_manager.clone(),
///     StreamManagerConfig {
///         max_concurrent_processors: 100,
///         log_operations: true,
///         global_flush_interval_secs: 5,
///     }
/// );
///
/// // Add streams, start processing, etc.
/// # Ok(())
/// # }
/// ```
pub struct StreamManager {
    /// Active stream configurations
    streams: Arc<RwLock<HashMap<Uuid, StreamConfig>>>,
    /// Active event subscriptions for streams
    subscriptions: Arc<RwLock<HashMap<Uuid, ListenerHandle>>>,
    /// Event manager for subscribing to events
    event_manager: Arc<EventManager>,
    /// Stream processors by stream id
    processors: ProcessorMap,
    /// Rate limiting semaphore for concurrent processing
    processing_semaphore: Arc<Semaphore>,
    /// Stream statistics
    stats: Arc<RwLock<HashMap<Uuid, StreamStats>>>,
    /// Configuration
    config: StreamManagerConfig,
    /// Long-lived per-stream event listener tasks
    listeners: TaskTracker,
    /// In-flight batch processing tasks
    batches: TaskTracker,
    /// Set to `true` by [`shutdown`](Self::shutdown); batches stop retrying when it flips.
    shutdown_signal: watch::Sender<bool>,
}

/// Configuration for the stream manager.
///
/// StreamManagerConfig controls the global behavior of the StreamManager,
/// including concurrency limits, logging, and flush intervals.
///
/// # Examples
///
/// ```rust
/// use hammerwork::streaming::StreamManagerConfig;
///
/// // Default configuration
/// let default_config = StreamManagerConfig::default();
///
/// // High-throughput configuration
/// let high_throughput = StreamManagerConfig {
///     max_concurrent_processors: 200,
///     log_operations: false, // Reduce logging overhead
///     global_flush_interval_secs: 30,
/// };
///
/// // Development configuration
/// let dev_config = StreamManagerConfig {
///     max_concurrent_processors: 10,
///     log_operations: true,
///     global_flush_interval_secs: 1, // Fast flushing for testing
/// };
/// ```
#[derive(Debug, Clone)]
pub struct StreamManagerConfig {
    /// Maximum number of batches being sent at once, across all streams. A batch
    /// holds a slot only while it is being sent, not while it waits to retry.
    pub max_concurrent_processors: usize,
    /// Whether to log stream operations
    pub log_operations: bool,
    /// Upper bound, in seconds, on how long any stream buffers an event before
    /// flushing it: a stream flushes after
    /// `min(buffer_config.max_buffer_time_secs, global_flush_interval_secs)`.
    pub global_flush_interval_secs: u64,
}

impl From<&crate::config::StreamingGlobalSettings> for StreamManagerConfig {
    fn from(settings: &crate::config::StreamingGlobalSettings) -> Self {
        Self {
            max_concurrent_processors: settings.max_concurrent_processors,
            log_operations: settings.log_operations,
            global_flush_interval_secs: settings.global_flush_interval_secs,
        }
    }
}

impl Default for StreamManagerConfig {
    fn default() -> Self {
        Self {
            max_concurrent_processors: 50,
            log_operations: true,
            global_flush_interval_secs: 10,
        }
    }
}

/// Trait for streaming backend processors.
///
/// StreamProcessor defines the interface that all streaming backend implementations
/// must implement. This allows the StreamManager to work with different streaming
/// systems in a uniform way.
///
/// # Implementation Requirements
///
/// Implementations must be thread-safe (`Send + Sync`) and handle:
/// - **Batch delivery** - Efficiently send multiple events in batches
/// - **Health monitoring** - Report backend health status
/// - **Error handling** - Properly handle and report delivery failures
/// - **Graceful shutdown** - Clean up resources when stopping
///
/// # Examples
///
/// ```rust
/// use hammerwork::streaming::{StreamProcessor, StreamedEvent, StreamDelivery};
/// use std::collections::HashMap;
/// use async_trait::async_trait;
///
/// struct MyCustomProcessor;
///
/// #[async_trait]
/// impl StreamProcessor for MyCustomProcessor {
///     async fn send_batch(&self, events: Vec<StreamedEvent>) -> hammerwork::Result<Vec<StreamDelivery>> {
///         // Implementation for sending events to custom backend
///         # Ok(Vec::new())
///     }
///
///     async fn health_check(&self) -> hammerwork::Result<bool> {
///         // Check if backend is healthy
///         Ok(true)
///     }
///
///     async fn get_stats(&self) -> hammerwork::Result<HashMap<String, serde_json::Value>> {
///         // Return processor-specific statistics
///         Ok(HashMap::new())
///     }
///
///     async fn shutdown(&self) -> hammerwork::Result<()> {
///         // Clean shutdown logic
///         Ok(())
///     }
/// }
/// ```
#[async_trait::async_trait]
pub trait StreamProcessor {
    /// Send a batch of events to the streaming backend.
    ///
    /// This method should efficiently deliver multiple events in a single operation
    /// to improve throughput. It returns a delivery result for each event.
    ///
    /// # Parameters
    ///
    /// * `events` - Batch of serialized events to deliver
    ///
    /// # Returns
    ///
    /// Vector of delivery results, one for each input event
    ///
    /// # Errors
    ///
    /// Returns an error if the entire batch fails to be processed
    async fn send_batch(&self, events: Vec<StreamedEvent>) -> crate::Result<Vec<StreamDelivery>>;

    /// Check if the processor is healthy and ready to send events.
    ///
    /// This method should perform a lightweight health check to determine
    /// if the streaming backend is available and ready to accept events.
    ///
    /// # Returns
    ///
    /// `Ok(true)` if healthy, `Ok(false)` if unhealthy, `Err` for check failures
    async fn health_check(&self) -> crate::Result<bool>;

    /// Get processor-specific statistics.
    ///
    /// Returns backend-specific metrics and statistics that can be used
    /// for monitoring and debugging.
    ///
    /// # Returns
    ///
    /// Map of statistic names to JSON values
    async fn get_stats(&self) -> crate::Result<HashMap<String, serde_json::Value>>;

    /// Shutdown the processor gracefully.
    ///
    /// This method should clean up any resources, close connections,
    /// and ensure no data is lost during shutdown.
    async fn shutdown(&self) -> crate::Result<()>;
}

/// Clamp a configured processor concurrency into the range `Semaphore` accepts.
///
/// `0` permits would make every processing task wait forever and more than
/// `Semaphore::MAX_PERMITS` panics inside tokio.
fn clamp_processor_permits(configured: usize) -> usize {
    clamp_permits(configured, "max_concurrent_processors")
}

/// First four characters of a credential for display, cut on a character
/// boundary (byte slicing would panic inside a multibyte character).
#[cfg_attr(not(feature = "kinesis"), allow(dead_code))]
fn mask_prefix(value: &str) -> String {
    value.chars().take(4).collect()
}

/// Read `health.check.timeout.ms`, defaulting to 5000 when unset but failing on
/// a value that is set and not a number (instead of silently using the default).
#[cfg_attr(
    not(any(feature = "kafka", feature = "google-pubsub")),
    allow(dead_code)
)]
fn parse_health_check_timeout_ms(config: &HashMap<String, String>) -> crate::Result<u64> {
    match config.get("health.check.timeout.ms") {
        None => Ok(5000),
        Some(value) => value.trim().parse::<u64>().map_err(|e| {
            crate::error::HammerworkError::Config(format!(
                "invalid health.check.timeout.ms '{}': {}",
                value, e
            ))
        }),
    }
}

/// Final result of delivering one batch, after retries.
#[derive(Debug, Default)]
struct BatchOutcome {
    /// Events the backend acknowledged.
    delivered: u64,
    /// Events that could not be prepared or were not delivered after all attempts.
    failed: u64,
    /// Sum of the delivery times of the delivered events, in milliseconds.
    delivery_time_ms_total: f64,
    /// The most recent error seen while delivering the batch.
    last_error: Option<String>,
}

impl StreamManager {
    /// Create a new stream manager.
    ///
    /// Creates a new StreamManager instance with the specified event manager
    /// and configuration. The stream manager will be ready to accept stream
    /// configurations and begin processing events.
    ///
    /// # Parameters
    ///
    /// * `event_manager` - Shared reference to the event manager for subscribing to events
    /// * `config` - Configuration for the stream manager behavior
    ///
    /// # Examples
    ///
    /// ```rust
    /// use hammerwork::streaming::{StreamManager, StreamManagerConfig};
    /// use hammerwork::events::EventManager;
    /// use std::sync::Arc;
    ///
    /// let event_manager = Arc::new(EventManager::new_default());
    /// let config = StreamManagerConfig {
    ///     max_concurrent_processors: 50,
    ///     log_operations: true,
    ///     global_flush_interval_secs: 10,
    /// };
    /// let stream_manager = StreamManager::new(event_manager, config);
    /// ```
    pub fn new(event_manager: Arc<EventManager>, config: StreamManagerConfig) -> Self {
        Self {
            streams: Arc::new(RwLock::new(HashMap::new())),
            subscriptions: Arc::new(RwLock::new(HashMap::new())),
            event_manager,
            processors: Arc::new(RwLock::new(HashMap::new())),
            processing_semaphore: Arc::new(Semaphore::new(clamp_processor_permits(
                config.max_concurrent_processors,
            ))),
            stats: Arc::new(RwLock::new(HashMap::new())),
            config,
            listeners: TaskTracker::new("stream listener"),
            batches: TaskTracker::new("stream batch"),
            shutdown_signal: watch::Sender::new(false),
        }
    }

    /// Stop the manager and wait for in-flight batches.
    ///
    /// Stops every stream's event listener, then waits up to `grace` for batches that
    /// are already being processed. A send that is in progress is allowed to finish,
    /// but no batch starts another retry after shutdown begins: events still awaiting
    /// a retry are counted as failed deliveries (with `last_error` saying why).
    /// Batches still running after `grace` are aborted. Finally every backend
    /// processor is shut down. Returns the number of batches aborted.
    ///
    /// Events still buffered in a listener (not yet flushed) are dropped.
    pub async fn shutdown(&self, grace: Duration) -> usize {
        self.shutdown_signal.send_replace(true);
        self.listeners.close();
        self.listeners.abort_all().await;
        let aborted = self.batches.drain(grace).await;

        let processors: Vec<_> = {
            let mut processors = self.processors.write().await;
            processors.drain().collect()
        };
        for (stream_id, processor) in processors {
            if let Err(e) = processor.shutdown().await {
                tracing::error!("Failed to shut down processor for stream {stream_id}: {e}");
            }
        }

        if self.config.log_operations {
            tracing::info!("Stream manager shut down ({aborted} batches aborted)");
        }
        aborted
    }

    /// Number of background stream tasks (listeners or batches) that have panicked.
    ///
    /// Panics are also logged at error level when they happen.
    pub fn task_panics(&self) -> u64 {
        self.listeners.panic_count() + self.batches.panic_count()
    }

    /// Create a new stream manager with default configuration.
    ///
    /// Convenience method that creates a StreamManager with default settings.
    /// Equivalent to calling `new()` with `StreamManagerConfig::default()`.
    ///
    /// # Parameters
    ///
    /// * `event_manager` - Shared reference to the event manager
    ///
    /// # Examples
    ///
    /// ```rust
    /// use hammerwork::streaming::StreamManager;
    /// use hammerwork::events::EventManager;
    /// use std::sync::Arc;
    ///
    /// let event_manager = Arc::new(EventManager::new_default());
    /// let stream_manager = StreamManager::new_default(event_manager);
    /// ```
    pub fn new_default(event_manager: Arc<EventManager>) -> Self {
        Self::new(event_manager, StreamManagerConfig::default())
    }

    /// Create a stream manager from the `streaming` section of a
    /// [`HammerworkConfig`](crate::HammerworkConfig), adding every configured stream.
    pub async fn from_config(
        event_manager: Arc<EventManager>,
        config: &crate::config::StreamingConfigs,
    ) -> crate::Result<Self> {
        let manager = Self::new(event_manager, (&config.global_settings).into());
        for stream in &config.streams {
            manager.add_stream(stream.clone()).await?;
        }
        Ok(manager)
    }

    /// Add a new stream configuration.
    ///
    /// Adds a new streaming configuration to the manager. This will:
    /// 1. Create the appropriate backend processor
    /// 2. Subscribe to events matching the stream's filter
    /// 3. Start processing events for this stream
    /// 4. Initialize statistics tracking
    ///
    /// # Parameters
    ///
    /// * `stream` - The stream configuration to add
    ///
    /// # Returns
    ///
    /// Returns `Ok(())` on success, or an error if the stream cannot be initialized
    ///
    /// # Examples
    ///
    /// ```rust
    /// use hammerwork::streaming::{StreamManager, StreamConfig, StreamBackend};
    /// use hammerwork::events::{EventManager, EventFilter};
    /// use std::sync::Arc;
    /// use std::collections::HashMap;
    ///
    /// # #[tokio::main]
    /// # async fn main() -> Result<(), Box<dyn std::error::Error>> {
    /// let event_manager = Arc::new(EventManager::new_default());
    /// let stream_manager = StreamManager::new_default(event_manager);
    ///
    /// let stream = StreamConfig {
    ///     id: uuid::Uuid::new_v4(),
    ///     name: "kafka-stream".to_string(),
    ///     backend: StreamBackend::Kafka {
    ///         brokers: vec!["localhost:9092".to_string()],
    ///         topic: "events".to_string(),
    ///         config: HashMap::new(),
    ///     },
    ///     filter: EventFilter::new(),
    ///     enabled: true,
    ///     ..Default::default()
    /// };
    ///
    /// stream_manager.add_stream(stream).await?;
    /// # Ok(())
    /// # }
    /// ```
    pub async fn add_stream(&self, stream: StreamConfig) -> crate::Result<()> {
        // Create processor for this stream
        let processor = self.create_processor(&stream).await?;
        self.add_stream_with_processor(stream, processor).await
    }

    /// Register a stream with an already-created processor.
    async fn add_stream_with_processor(
        &self,
        stream: StreamConfig,
        processor: Box<dyn StreamProcessor + Send + Sync>,
    ) -> crate::Result<()> {
        let stream_id = stream.id;

        // Subscribe before returning, and hand this receiver to the listener: events
        // published from now on are buffered for it even before its task first runs.
        let subscription = self.event_manager.subscribe(stream.filter.clone()).await?;
        let (handle, stop) = ListenerHandle::new(subscription.id);

        // Store stream and processor
        {
            let mut streams = self.streams.write().await;
            streams.insert(stream_id, stream.clone());
        }

        {
            let mut processors = self.processors.write().await;
            processors.insert(stream_id, Arc::from(processor));
        }

        // Initialize statistics
        {
            let mut stats = self.stats.write().await;
            stats.insert(
                stream_id,
                StreamStats {
                    stream_id,
                    total_events: 0,
                    successful_deliveries: 0,
                    failed_deliveries: 0,
                    success_rate: 0.0,
                    avg_delivery_time_ms: 0.0,
                    buffered_events: 0,
                    last_success_at: None,
                    last_failure_at: None,
                    last_error: None,
                    dropped_events: 0,
                    calculated_at: Utc::now(),
                },
            );
        }

        // Replacing the handle of an existing listener (same stream id) stops it.
        let replaced = self.subscriptions.write().await.insert(stream_id, handle);
        if let Some(replaced) = replaced {
            self.event_manager
                .unsubscribe(replaced.subscription_id)
                .await?;
        }

        // Start processing task for this stream
        self.start_stream_processing_task(stream_id, subscription.receiver, stop);

        if self.config.log_operations {
            tracing::info!("Added stream: {} ({:?})", stream.name, stream.backend);
        }

        Ok(())
    }

    /// Test helper: add a stream, using an in-memory processor for backends
    /// whose cargo feature is disabled (their real constructors fail fast).
    /// Backends that are compiled in still use their real processor.
    #[cfg(test)]
    pub(crate) async fn add_stream_for_test(&self, stream: StreamConfig) -> crate::Result<()> {
        let backend_compiled = match &stream.backend {
            StreamBackend::Kafka { .. } => cfg!(feature = "kafka"),
            StreamBackend::Kinesis { .. } => cfg!(feature = "kinesis"),
            StreamBackend::PubSub { .. } => cfg!(feature = "google-pubsub"),
        };
        if backend_compiled {
            self.add_stream(stream).await
        } else {
            self.add_stream_with_processor(stream, Box::new(InMemoryProcessor::default()))
                .await
        }
    }

    /// Remove a stream configuration
    pub async fn remove_stream(&self, stream_id: Uuid) -> crate::Result<()> {
        // Stop the listener first (dropping its handle) and remove the subscription
        let handle = self.subscriptions.write().await.remove(&stream_id);
        if let Some(handle) = handle {
            self.event_manager
                .unsubscribe(handle.subscription_id)
                .await?;
        }

        // Shutdown processor
        let processor = self.processors.write().await.remove(&stream_id);
        if let Some(processor) = processor {
            processor.shutdown().await?;
        }

        // Remove stream configuration
        {
            let mut streams = self.streams.write().await;
            streams.remove(&stream_id);
        }

        // Remove statistics
        {
            let mut stats = self.stats.write().await;
            stats.remove(&stream_id);
        }

        if self.config.log_operations {
            tracing::info!("Removed stream: {}", stream_id);
        }

        Ok(())
    }

    /// Get stream configuration
    pub async fn get_stream(&self, stream_id: Uuid) -> Option<StreamConfig> {
        let streams = self.streams.read().await;
        streams.get(&stream_id).cloned()
    }

    /// List all stream configurations
    pub async fn list_streams(&self) -> Vec<StreamConfig> {
        let streams = self.streams.read().await;
        streams.values().cloned().collect()
    }

    /// Enable a stream
    pub async fn enable_stream(&self, stream_id: Uuid) -> crate::Result<()> {
        let mut streams = self.streams.write().await;
        if let Some(stream) = streams.get_mut(&stream_id) {
            stream.enabled = true;
            Ok(())
        } else {
            Err(crate::HammerworkError::Queue {
                message: format!("Stream {} not found", stream_id),
            })
        }
    }

    /// Disable a stream  
    pub async fn disable_stream(&self, stream_id: Uuid) -> crate::Result<()> {
        let mut streams = self.streams.write().await;
        if let Some(stream) = streams.get_mut(&stream_id) {
            stream.enabled = false;
            Ok(())
        } else {
            Err(crate::HammerworkError::Queue {
                message: format!("Stream {} not found", stream_id),
            })
        }
    }

    /// Get stream statistics
    pub async fn get_stream_stats(&self, stream_id: Uuid) -> Option<StreamStats> {
        let stats = self.stats.read().await;
        stats.get(&stream_id).cloned()
    }

    /// Get statistics for all streams
    pub async fn get_all_stream_stats(&self) -> Vec<StreamStats> {
        let stats = self.stats.read().await;
        stats.values().cloned().collect()
    }

    /// Get general stream manager statistics
    pub async fn get_stats(&self) -> StreamManagerGlobalStats {
        let streams = self.streams.read().await;
        let stats = self.stats.read().await;

        let total_streams = streams.len();
        let active_streams = streams.values().filter(|s| s.enabled).count();
        let total_events = stats.values().map(|s| s.total_events).sum();
        let successful_deliveries = stats.values().map(|s| s.successful_deliveries).sum();
        let failed_deliveries = stats.values().map(|s| s.failed_deliveries).sum();

        StreamManagerGlobalStats {
            total_streams,
            active_streams,
            total_events,
            successful_deliveries,
            failed_deliveries,
        }
    }

    /// Create a processor for the given stream configuration
    async fn create_processor(
        &self,
        stream: &StreamConfig,
    ) -> crate::Result<Box<dyn StreamProcessor + Send + Sync>> {
        match &stream.backend {
            StreamBackend::Kafka {
                brokers,
                topic,
                config,
            } => Ok(Box::new(
                KafkaProcessor::new(brokers.clone(), topic.clone(), config.clone()).await?,
            )),
            StreamBackend::Kinesis {
                region,
                stream_name,
                access_key_id,
                secret_access_key,
                config,
            } => Ok(Box::new(
                KinesisProcessor::new(
                    region.clone(),
                    stream_name.clone(),
                    access_key_id.clone(),
                    secret_access_key.clone(),
                    config.clone(),
                )
                .await?,
            )),
            StreamBackend::PubSub {
                project_id,
                topic_name,
                service_account_key,
                config,
            } => Ok(Box::new(
                PubSubProcessor::new(
                    project_id.clone(),
                    topic_name.clone(),
                    service_account_key.clone(),
                    config.clone(),
                )
                .await?,
            )),
        }
    }

    /// Start the listener of a stream: it reads `receiver` until `stop` fires (its
    /// [`ListenerHandle`] is dropped), buffers matching events and hands full or
    /// expired buffers to batch tasks.
    ///
    /// A buffer is flushed when it holds `batch_size` or `max_events` events, or when
    /// its oldest event has waited [`flush_after`](Self::flush_after). At most
    /// `max_events` events are in batches at once (being sent or waiting to retry);
    /// beyond that the listener waits, and events that overflow the event manager's
    /// buffer meanwhile are dropped and counted in [`StreamStats::dropped_events`].
    fn start_stream_processing_task(
        &self,
        stream_id: Uuid,
        mut receiver: broadcast::Receiver<JobLifecycleEvent>,
        mut stop: oneshot::Receiver<()>,
    ) {
        let streams = self.streams.clone();
        let processors = self.processors.clone();
        let stats = self.stats.clone();
        let processing_semaphore = self.processing_semaphore.clone();
        let config = self.config.clone();
        let shutdown_signal = self.shutdown_signal.clone();
        let batches = self.batches.clone();

        self.listeners.spawn("listener", async move {
            let mut event_buffer: Vec<JobLifecycleEvent> = Vec::new();
            // When the oldest buffered event arrived.
            let mut buffered_since: Option<tokio::time::Instant> = None;
            // Events handed to batches and not finished yet.
            let mut in_flight_limit: Option<(usize, Arc<Semaphore>)> = None;
            // Whether the listener is waiting for in-flight batches (logged once per pause).
            let mut paused = false;

            loop {
                // Get stream configuration
                let stream = {
                    let streams = streams.read().await;
                    match streams.get(&stream_id) {
                        Some(stream) => stream.clone(),
                        // Stream removed, exit task
                        None => break,
                    }
                };
                let max_events = clamp_permits(stream.buffer_config.max_events, "max_events");
                let in_flight = match &in_flight_limit {
                    Some((limit, semaphore)) if *limit == max_events => semaphore.clone(),
                    _ => {
                        let semaphore = Arc::new(Semaphore::new(max_events));
                        in_flight_limit = Some((max_events, semaphore.clone()));
                        semaphore
                    }
                };
                let flush_after = Self::flush_after(&stream, &config);
                let flush_at = buffered_since.map(|since| since + flush_after);

                let should_flush = event_buffer.len() >= stream.buffer_config.batch_size
                    || event_buffer.len() >= max_events
                    || flush_at.is_some_and(|at| tokio::time::Instant::now() >= at);

                if should_flush && !event_buffer.is_empty() {
                    let events_to_process = std::mem::take(&mut event_buffer);
                    buffered_since = None;

                    // Bound the backlog: wait until these events fit in the in-flight
                    // limit. While waiting, new events queue up in the broadcast buffer.
                    let wanted =
                        u32::try_from(events_to_process.len().min(max_events)).unwrap_or(u32::MAX);
                    let full = (in_flight.available_permits() as u32) < wanted;
                    if full && !paused && config.log_operations {
                        tracing::warn!(
                            "Stream {} has {} events in flight; pausing its listener",
                            stream.name,
                            max_events - in_flight.available_permits()
                        );
                    }
                    paused = full;
                    let slots = tokio::select! {
                        biased;
                        _ = &mut stop => break,
                        slots = in_flight.clone().acquire_many_owned(wanted) => match slots {
                            Ok(slots) => slots,
                            Err(_) => break,
                        },
                    };

                    let stream_clone = stream.clone();
                    let processors_clone = processors.clone();
                    let stats_clone = stats.clone();
                    let config_clone = config.clone();
                    let semaphore_clone = processing_semaphore.clone();
                    let shutdown = shutdown_signal.subscribe();

                    batches.spawn("batch", async move {
                        Self::process_event_batch(
                            stream_id,
                            stream_clone,
                            events_to_process,
                            processors_clone,
                            stats_clone,
                            config_clone,
                            semaphore_clone,
                            shutdown,
                            slots,
                        )
                        .await;
                    });
                    continue;
                }

                // Wait for the next event, or until the buffer is due to be flushed.
                let flush_due = async {
                    match flush_at {
                        Some(at) => tokio::time::sleep_until(at).await,
                        None => std::future::pending().await,
                    }
                };
                let received = tokio::select! {
                    biased;
                    _ = &mut stop => break,
                    received = receiver.recv() => received,
                    _ = flush_due => continue,
                };
                match received {
                    Ok(event) => {
                        // A disabled stream drops events (but keeps listening, so
                        // that enabling it again resumes delivery). Check the current
                        // state: it may have changed while waiting.
                        let enabled = streams
                            .read()
                            .await
                            .get(&stream_id)
                            .is_some_and(|stream| stream.enabled);
                        if enabled && stream.filter.matches(&event) {
                            if event_buffer.is_empty() {
                                buffered_since = Some(tokio::time::Instant::now());
                            }
                            event_buffer.push(event);
                        }
                    }
                    Err(broadcast::error::RecvError::Lagged(skipped)) => {
                        tracing::warn!(
                            "Stream {} listener fell behind and dropped {} events; the \
                             backend may be slow or down, or the event buffer too small",
                            stream.name,
                            skipped
                        );
                        if let Some(stream_stats) = stats.write().await.get_mut(&stream_id) {
                            stream_stats.dropped_events += skipped;
                        }
                    }
                    Err(broadcast::error::RecvError::Closed) => {
                        // Event manager dropped, exit task
                        break;
                    }
                }
            }
        });
    }

    /// How long an event may wait in a stream's buffer: the stream's
    /// `max_buffer_time_secs`, capped by the manager's `global_flush_interval_secs`.
    fn flush_after(stream: &StreamConfig, config: &StreamManagerConfig) -> Duration {
        Duration::from_secs(
            stream
                .buffer_config
                .max_buffer_time_secs
                .min(config.global_flush_interval_secs),
        )
    }

    /// Deliver a batch of events through the stream's processor.
    ///
    /// Events are serialized, sent with [`StreamProcessor::send_batch`], and every
    /// event whose delivery failed (or that has no delivery result) is retried
    /// according to the stream's [`StreamRetryPolicy`]. Statistics are updated from
    /// the final per-event outcome.
    ///
    /// `processing_semaphore` limits sends across all streams; a send holds a slot only
    /// while it runs, not during the backoff before a retry. `_in_flight` is the
    /// stream's in-flight allowance for these events, released when the batch is done.
    #[allow(clippy::too_many_arguments)]
    async fn process_event_batch(
        stream_id: Uuid,
        stream: StreamConfig,
        events: Vec<JobLifecycleEvent>,
        processors: ProcessorMap,
        stats: Arc<RwLock<HashMap<Uuid, StreamStats>>>,
        config: StreamManagerConfig,
        processing_semaphore: Arc<Semaphore>,
        mut shutdown: watch::Receiver<bool>,
        _in_flight: OwnedSemaphorePermit,
    ) {
        // The stream may have been removed while the batch waited.
        let Some(processor) = processors.read().await.get(&stream_id).cloned() else {
            return;
        };

        let mut outcome = BatchOutcome::default();

        // Convert events to streamed events
        let mut streamed_events = Vec::with_capacity(events.len());
        for event in events {
            match Self::prepare_streamed_event(event, &stream) {
                Ok(streamed_event) => streamed_events.push(streamed_event),
                Err(e) => {
                    let message = format!("failed to prepare event for streaming: {e}");
                    if config.log_operations {
                        tracing::error!("Stream {}: {}", stream.name, message);
                    }
                    outcome.failed += 1;
                    outcome.last_error = Some(message);
                }
            }
        }

        if !streamed_events.is_empty() {
            if config.log_operations {
                tracing::debug!(
                    "Processing batch of {} events for stream {}",
                    streamed_events.len(),
                    stream.name
                );
            }
            Self::deliver_with_retries(
                &stream,
                processor.as_ref(),
                streamed_events,
                &processing_semaphore,
                &mut shutdown,
                &config,
                &mut outcome,
            )
            .await;
        }

        Self::record_batch_outcome(stream_id, &outcome, &stats).await;
    }

    /// Send `pending` through `processor`, retrying failed events per the stream's
    /// retry policy until they are delivered, the attempts run out, or the manager
    /// shuts down. Results are accumulated into `outcome`.
    async fn deliver_with_retries(
        stream: &StreamConfig,
        processor: &(dyn StreamProcessor + Send + Sync),
        mut pending: Vec<StreamedEvent>,
        processing_semaphore: &Semaphore,
        shutdown: &mut watch::Receiver<bool>,
        config: &StreamManagerConfig,
        outcome: &mut BatchOutcome,
    ) {
        let policy = &stream.retry_policy;
        let max_attempts = policy.max_attempts.max(1);
        let mut attempt = 0u32;

        loop {
            attempt += 1;
            let last_attempt = attempt >= max_attempts;
            let event_ids: Vec<Uuid> = pending.iter().map(|e| e.event.event_id).collect();
            // Keep a copy for the next retry; the final attempt can hand the events over.
            let batch = if last_attempt {
                std::mem::take(&mut pending)
            } else {
                pending.clone()
            };

            let Ok(permit) = processing_semaphore.acquire().await else {
                tracing::error!("stream processing semaphore closed");
                outcome.failed += event_ids.len() as u64;
                return;
            };
            let started = std::time::Instant::now();
            let result = processor.send_batch(batch).await;
            drop(permit);
            let elapsed_ms = u64::try_from(started.elapsed().as_millis()).unwrap_or(u64::MAX);

            let failed_ids: HashSet<Uuid> = match result {
                Ok(deliveries) => {
                    let by_event: HashMap<Uuid, StreamDelivery> = deliveries
                        .into_iter()
                        .map(|delivery| (delivery.event_id, delivery))
                        .collect();
                    let mut failed = HashSet::new();
                    for event_id in &event_ids {
                        match by_event.get(event_id) {
                            Some(delivery) if delivery.success => {
                                outcome.delivered += 1;
                                outcome.delivery_time_ms_total +=
                                    delivery.duration_ms.unwrap_or(elapsed_ms) as f64;
                            }
                            Some(delivery) => {
                                outcome.last_error = Some(
                                    delivery
                                        .error_message
                                        .clone()
                                        .unwrap_or_else(|| "delivery failed".to_string()),
                                );
                                failed.insert(*event_id);
                            }
                            None => {
                                outcome.last_error =
                                    Some("processor returned no delivery result".to_string());
                                failed.insert(*event_id);
                            }
                        }
                    }
                    failed
                }
                Err(e) => {
                    outcome.last_error = Some(e.to_string());
                    event_ids.iter().copied().collect()
                }
            };

            if failed_ids.is_empty() {
                return;
            }
            if last_attempt {
                outcome.failed += failed_ids.len() as u64;
                if config.log_operations {
                    tracing::error!(
                        "Stream {}: {} event(s) failed after {} attempt(s): {}",
                        stream.name,
                        failed_ids.len(),
                        attempt,
                        outcome.last_error.as_deref().unwrap_or("unknown error")
                    );
                }
                return;
            }

            pending.retain(|e| failed_ids.contains(&e.event.event_id));
            let delay = policy.delay_for_retry(attempt);
            if config.log_operations {
                tracing::warn!(
                    "Stream {}: {} event(s) failed (attempt {}/{}), retrying in {:?}: {}",
                    stream.name,
                    pending.len(),
                    attempt,
                    max_attempts,
                    delay,
                    outcome.last_error.as_deref().unwrap_or("unknown error")
                );
            }

            // Stop retrying once shutdown starts, including during the backoff sleep.
            let shutting_down = *shutdown.borrow()
                || tokio::select! {
                    _ = tokio::time::sleep(delay) => false,
                    _ = shutdown.wait_for(|stopping| *stopping) => true,
                };
            if shutting_down {
                outcome.failed += pending.len() as u64;
                let reason = format!(
                    "stream manager shut down before retry {} of {}; last error: {}",
                    attempt + 1,
                    max_attempts,
                    outcome.last_error.as_deref().unwrap_or("unknown error")
                );
                if config.log_operations {
                    tracing::warn!(
                        "Stream {}: giving up on {} event(s): {}",
                        stream.name,
                        pending.len(),
                        reason
                    );
                }
                outcome.last_error = Some(reason);
                return;
            }
        }
    }

    /// Prepare a job lifecycle event for streaming
    fn prepare_streamed_event(
        event: JobLifecycleEvent,
        stream: &StreamConfig,
    ) -> crate::Result<StreamedEvent> {
        // Calculate partition key
        let partition_key = Self::calculate_partition_key(&event, &stream.partitioning);

        // Serialize event data
        let serialized_data = match &stream.serialization {
            SerializationFormat::Json => serde_json::to_vec(&event)?,
            #[cfg(feature = "streaming")]
            SerializationFormat::MessagePack => {
                rmp_serde::to_vec(&event).map_err(|e| HammerworkError::Streaming {
                    message: format!("MessagePack serialization failed: {}", e),
                })?
            }
            #[cfg(not(feature = "streaming"))]
            SerializationFormat::MessagePack => {
                return Err(HammerworkError::Streaming {
                    message: "MessagePack serialization requires 'streaming' feature".to_string(),
                });
            }
            #[cfg(feature = "streaming")]
            SerializationFormat::Avro {
                schema_registry_url,
            } => Self::serialize_avro(&event, schema_registry_url)?,
            #[cfg(not(feature = "streaming"))]
            SerializationFormat::Avro { .. } => {
                return Err(HammerworkError::Streaming {
                    message: "Avro serialization requires 'streaming' feature".to_string(),
                });
            }
            #[cfg(feature = "streaming")]
            SerializationFormat::Protobuf { schema_definition } => {
                Self::serialize_protobuf(&event, schema_definition)?
            }
            #[cfg(not(feature = "streaming"))]
            SerializationFormat::Protobuf { .. } => {
                return Err(HammerworkError::Streaming {
                    message: "Protobuf serialization requires 'streaming' feature".to_string(),
                });
            }
        };

        // Prepare headers
        let mut headers = HashMap::new();
        headers.insert("event_type".to_string(), event.event_type.to_string());
        headers.insert("queue_name".to_string(), event.queue_name.clone());
        headers.insert("priority".to_string(), event.priority.to_string());
        headers.insert("timestamp".to_string(), event.timestamp.to_rfc3339());

        Ok(StreamedEvent {
            event,
            partition_key,
            serialized_data,
            streamed_at: Utc::now(),
            headers,
        })
    }

    #[cfg(feature = "streaming")]
    /// Serialize event data using Avro format
    fn serialize_avro(
        event: &JobLifecycleEvent,
        _schema_registry_url: &str,
    ) -> crate::Result<Vec<u8>> {
        use apache_avro::{Schema, Writer};

        // Define Avro schema for JobLifecycleEvent
        let schema = Schema::parse_str(
            r#"
        {
            "type": "record",
            "name": "JobLifecycleEvent",
            "fields": [
                {"name": "job_id", "type": "string"},
                {"name": "queue_name", "type": "string"},
                {"name": "event_type", "type": "string"},
                {"name": "priority", "type": "string"},
                {"name": "timestamp", "type": "string"},
                {"name": "metadata", "type": {"type": "map", "values": "string"}}
            ]
        }
        "#,
        )
        .map_err(|e| HammerworkError::Streaming {
            message: format!("Failed to parse Avro schema: {}", e),
        })?;

        let mut writer = Writer::new(&schema, Vec::new());

        // Create Avro record
        let mut record =
            apache_avro::types::Record::new(&schema).ok_or_else(|| HammerworkError::Streaming {
                message: "Failed to create Avro record".to_string(),
            })?;

        record.put("job_id", event.job_id.to_string());
        record.put("queue_name", event.queue_name.clone());
        record.put("event_type", event.event_type.to_string());
        record.put("priority", event.priority.to_string());
        record.put("timestamp", event.timestamp.to_rfc3339());
        record.put("metadata", event.metadata.clone());

        writer
            .append(record)
            .map_err(|e| HammerworkError::Streaming {
                message: format!("Failed to append Avro record: {}", e),
            })?;

        writer.into_inner().map_err(|e| HammerworkError::Streaming {
            message: format!("Failed to finalize Avro writer: {}", e),
        })
    }

    #[cfg(feature = "streaming")]
    /// Serialize event data using Protobuf format
    fn serialize_protobuf(
        event: &JobLifecycleEvent,
        _schema_definition: &str,
    ) -> crate::Result<Vec<u8>> {
        use prost::Message;

        // Define protobuf message structure
        #[derive(prost::Message)]
        struct JobLifecycleEventProto {
            #[prost(string, tag = "1")]
            job_id: String,
            #[prost(string, tag = "2")]
            queue_name: String,
            #[prost(string, tag = "3")]
            event_type: String,
            #[prost(string, tag = "4")]
            priority: String,
            #[prost(string, tag = "5")]
            timestamp: String,
            #[prost(map = "string, string", tag = "6")]
            metadata: std::collections::HashMap<String, String>,
        }

        let proto_event = JobLifecycleEventProto {
            job_id: event.job_id.to_string(),
            queue_name: event.queue_name.clone(),
            event_type: event.event_type.to_string(),
            priority: event.priority.to_string(),
            timestamp: event.timestamp.to_rfc3339(),
            metadata: event.metadata.clone(),
        };

        let mut buf = Vec::new();
        proto_event
            .encode(&mut buf)
            .map_err(|e| HammerworkError::Streaming {
                message: format!("Failed to encode Protobuf message: {}", e),
            })?;

        Ok(buf)
    }

    /// Calculate partition key based on partitioning strategy
    fn calculate_partition_key(
        event: &JobLifecycleEvent,
        strategy: &PartitioningStrategy,
    ) -> Option<String> {
        match strategy {
            PartitioningStrategy::None => None,
            PartitioningStrategy::JobId => Some(event.job_id.to_string()),
            PartitioningStrategy::QueueName => Some(event.queue_name.clone()),
            PartitioningStrategy::Priority => Some(event.priority.to_string()),
            PartitioningStrategy::EventType => Some(event.event_type.to_string()),
            PartitioningStrategy::Custom { metadata_key } => {
                event.metadata.get(metadata_key).cloned()
            }
            PartitioningStrategy::Hash { fields } => {
                let mut hash_input = String::new();
                for field in fields {
                    match field {
                        PartitionField::JobId => hash_input.push_str(&event.job_id.to_string()),
                        PartitionField::QueueName => hash_input.push_str(&event.queue_name),
                        PartitionField::Priority => {
                            hash_input.push_str(&event.priority.to_string())
                        }
                        PartitionField::EventType => {
                            hash_input.push_str(&event.event_type.to_string())
                        }
                        PartitionField::MetadataKey(key) => {
                            if let Some(value) = event.metadata.get(key) {
                                hash_input.push_str(value);
                            }
                        }
                    }
                    hash_input.push('|');
                }

                // Simple hash function (in production, use a proper hash function)
                let hash = hash_input.chars().map(|c| c as u32).sum::<u32>();
                Some(format!("{}", hash % 1000))
            }
        }
    }

    /// Fold the final outcome of one batch into the stream's statistics.
    async fn record_batch_outcome(
        stream_id: Uuid,
        outcome: &BatchOutcome,
        stats: &RwLock<HashMap<Uuid, StreamStats>>,
    ) {
        let mut stats_map = stats.write().await;
        let Some(stream_stats) = stats_map.get_mut(&stream_id) else {
            return;
        };
        let now = Utc::now();
        stream_stats.total_events += outcome.delivered + outcome.failed;

        if outcome.delivered > 0 {
            let previous = stream_stats.successful_deliveries as f64;
            stream_stats.successful_deliveries += outcome.delivered;
            stream_stats.avg_delivery_time_ms = (stream_stats.avg_delivery_time_ms * previous
                + outcome.delivery_time_ms_total)
                / stream_stats.successful_deliveries as f64;
            stream_stats.last_success_at = Some(now);
        }
        if outcome.failed > 0 {
            stream_stats.failed_deliveries += outcome.failed;
            stream_stats.last_failure_at = Some(now);
            stream_stats.last_error = outcome.last_error.clone();
        }

        let finished = stream_stats.successful_deliveries + stream_stats.failed_deliveries;
        if finished > 0 {
            stream_stats.success_rate = stream_stats.successful_deliveries as f64 / finished as f64;
        }
        stream_stats.calculated_at = now;
    }
}

#[cfg(feature = "kafka")]
/// Kafka processor with real rdkafka implementation
pub struct KafkaProcessor {
    producer: FutureProducer,
    topic: String,
    config: HashMap<String, String>,
}

#[cfg(feature = "kafka")]
impl KafkaProcessor {
    pub async fn new(
        brokers: Vec<String>,
        topic: String,
        config: HashMap<String, String>,
    ) -> crate::Result<Self> {
        let mut client_config = ClientConfig::new();

        // Set basic configuration
        client_config.set("bootstrap.servers", brokers.join(","));

        // Apply custom configuration. `health.check.timeout.ms` is read by
        // `health_check` and is not a librdkafka property, so it is not passed on.
        for (key, value) in &config {
            if key != "health.check.timeout.ms" {
                client_config.set(key, value);
            }
        }

        // Set defaults if not provided
        if !config.contains_key("message.timeout.ms") {
            client_config.set("message.timeout.ms", "5000");
        }
        if !config.contains_key("queue.buffering.max.messages") {
            client_config.set("queue.buffering.max.messages", "100000");
        }

        let producer: FutureProducer =
            client_config
                .create()
                .map_err(|e| HammerworkError::Streaming {
                    message: format!("Failed to create Kafka producer: {}", e),
                })?;

        Ok(Self {
            producer,
            topic,
            config,
        })
    }
}

#[cfg(feature = "kafka")]
#[async_trait::async_trait]
impl StreamProcessor for KafkaProcessor {
    async fn send_batch(&self, events: Vec<StreamedEvent>) -> crate::Result<Vec<StreamDelivery>> {
        let mut deliveries = Vec::new();

        // Send events to Kafka
        for event in events {
            let delivery_start = Utc::now();

            // Create Kafka record
            let mut record = FutureRecord::to(&self.topic).payload(&event.serialized_data);

            // Add partition key if available
            if let Some(ref key) = event.partition_key {
                record = record.key(key);
            }

            // Add headers
            if !event.headers.is_empty() {
                let mut headers = OwnedHeaders::new();
                for (header_key, header_value) in &event.headers {
                    headers = headers.insert(Header {
                        key: header_key,
                        value: Some(header_value),
                    });
                }
                record = record.headers(headers);
            }

            // Send the message
            let result = self
                .producer
                .send(record, tokio::time::Duration::from_secs(5))
                .await;

            let delivery_end = Utc::now();
            let duration_ms = delivery_end
                .signed_duration_since(delivery_start)
                .num_milliseconds()
                .max(0) as u64;

            let (success, error_message, partition) = match result {
                Ok((partition, offset)) => {
                    tracing::debug!(
                        "Message delivered to partition {} at offset {}",
                        partition,
                        offset
                    );
                    (true, None, Some(partition.to_string()))
                }
                Err((kafka_error, _)) => {
                    let error_msg = format!("Kafka delivery failed: {}", kafka_error);
                    tracing::error!("{}", error_msg);
                    (false, Some(error_msg), event.partition_key)
                }
            };

            deliveries.push(StreamDelivery {
                delivery_id: Uuid::new_v4(),
                stream_id: Uuid::new_v4(), // Would be passed in from context
                event_id: event.event.event_id,
                success,
                error_message,
                attempted_at: delivery_start,
                duration_ms: Some(duration_ms),
                attempt_number: 1,
                partition,
            });
        }

        Ok(deliveries)
    }

    async fn health_check(&self) -> crate::Result<bool> {
        let health_check_timeout_ms = parse_health_check_timeout_ms(&self.config)?;

        // Get cluster metadata to verify connectivity
        let timeout = tokio::time::Duration::from_millis(health_check_timeout_ms);

        // Try to get metadata (doesn't actually send the message)
        let result = tokio::task::spawn_blocking({
            let producer = self.producer.clone();
            let topic = self.topic.clone();
            move || producer.client().fetch_metadata(Some(&topic), timeout)
        })
        .await;

        let health_result = match result {
            Ok(metadata_result) => match metadata_result {
                Ok(metadata) => {
                    let topic_metadata = metadata.topics().iter().find(|t| t.name() == self.topic);

                    match topic_metadata {
                        Some(topic) => {
                            if topic.partitions().is_empty() {
                                tracing::warn!("Topic {} has no partitions", self.topic);
                                false
                            } else {
                                tracing::debug!(
                                    "Kafka health check passed for topic {}",
                                    self.topic
                                );
                                true
                            }
                        }
                        None => {
                            tracing::warn!("Topic {} not found in metadata", self.topic);
                            false
                        }
                    }
                }
                Err(e) => {
                    tracing::error!("Kafka health check failed: {}", e);
                    false
                }
            },
            Err(e) => {
                tracing::error!("Kafka health check task failed: {}", e);
                false
            }
        };

        Ok(health_result)
    }

    async fn get_stats(&self) -> crate::Result<HashMap<String, serde_json::Value>> {
        let mut stats = HashMap::new();
        stats.insert(
            "type".to_string(),
            serde_json::Value::String("kafka".to_string()),
        );

        // Get broker information from producer configuration
        let brokers = self
            .config
            .get("bootstrap.servers")
            .map(|b| {
                b.split(',')
                    .map(|s| s.trim().to_string())
                    .collect::<Vec<_>>()
            })
            .unwrap_or_default();

        stats.insert(
            "brokers".to_string(),
            serde_json::Value::Array(
                brokers
                    .iter()
                    .map(|b| serde_json::Value::String(b.clone()))
                    .collect(),
            ),
        );
        stats.insert(
            "topic".to_string(),
            serde_json::Value::String(self.topic.clone()),
        );

        // Include basic producer statistics (if available)
        // Note: rdkafka statistics might not be available in all configurations
        stats.insert(
            "producer_available".to_string(),
            serde_json::Value::Bool(true),
        );

        stats.insert(
            "config".to_string(),
            serde_json::Value::Object(
                self.config
                    .iter()
                    .map(|(k, v)| (k.clone(), serde_json::Value::String(v.clone())))
                    .collect(),
            ),
        );
        Ok(stats)
    }

    async fn shutdown(&self) -> crate::Result<()> {
        // Flush any pending messages and gracefully shutdown
        let producer = self.producer.clone();
        let flush_result = tokio::task::spawn_blocking(move || {
            producer.flush(tokio::time::Duration::from_secs(10))
        })
        .await;

        match flush_result {
            Ok(Ok(())) => {
                tracing::debug!("Kafka producer flushed successfully during shutdown");
            }
            Ok(Err(e)) => {
                tracing::warn!("Error flushing Kafka producer during shutdown: {}", e);
            }
            Err(e) => {
                tracing::warn!("Flush task failed during shutdown: {}", e);
            }
        }
        Ok(())
    }
}

#[cfg(not(feature = "kafka"))]
/// Uninhabited stand-in compiled when the `kafka` feature is disabled.
///
/// [`KafkaProcessor::new`] always fails, so a value of this type can never exist and
/// streams for this backend are rejected when they are configured.
pub struct KafkaProcessor(std::convert::Infallible);

#[cfg(not(feature = "kafka"))]
impl KafkaProcessor {
    /// Always fails: Kafka support requires the `kafka` feature.
    pub async fn new(
        _brokers: Vec<String>,
        _topic: String,
        _config: HashMap<String, String>,
    ) -> crate::Result<Self> {
        Err(HammerworkError::Streaming {
            message: "Kafka support requires the `kafka` feature of hammerwork".to_string(),
        })
    }
}

#[cfg(not(feature = "kafka"))]
#[async_trait::async_trait]
impl StreamProcessor for KafkaProcessor {
    async fn send_batch(&self, _events: Vec<StreamedEvent>) -> crate::Result<Vec<StreamDelivery>> {
        match self.0 {}
    }

    async fn health_check(&self) -> crate::Result<bool> {
        match self.0 {}
    }

    async fn get_stats(&self) -> crate::Result<HashMap<String, serde_json::Value>> {
        match self.0 {}
    }

    async fn shutdown(&self) -> crate::Result<()> {
        match self.0 {}
    }
}

/// AWS Kinesis processor for streaming job events
#[cfg(feature = "kinesis")]
pub struct KinesisProcessor {
    region: String,
    stream_name: String,
    access_key_id: Option<String>,
    secret_access_key: Option<String>,
    config: HashMap<String, String>,
    client: KinesisClient,
}

#[cfg(feature = "kinesis")]
impl KinesisProcessor {
    pub async fn new(
        region: String,
        stream_name: String,
        access_key_id: Option<String>,
        secret_access_key: Option<String>,
        config: HashMap<String, String>,
    ) -> crate::Result<Self> {
        // Initialize AWS configuration
        let aws_config =
            if let (Some(access_key), Some(secret_key)) = (&access_key_id, &secret_access_key) {
                // Use provided credentials
                let credentials = aws_sdk_kinesis::config::Credentials::new(
                    access_key,
                    secret_key,
                    None, // session token
                    None, // expiry
                    "hammerwork",
                );

                aws_config::defaults(BehaviorVersion::latest())
                    .region(Region::new(region.clone()))
                    .credentials_provider(credentials)
                    .load()
                    .await
            } else {
                // Use default credential chain (IAM roles, environment variables, etc.)
                aws_config::defaults(BehaviorVersion::latest())
                    .region(Region::new(region.clone()))
                    .load()
                    .await
            };

        let client = KinesisClient::new(&aws_config);

        Ok(Self {
            region,
            stream_name,
            access_key_id,
            secret_access_key,
            config,
            client,
        })
    }
}

#[cfg(feature = "kinesis")]
#[async_trait::async_trait]
impl StreamProcessor for KinesisProcessor {
    async fn send_batch(&self, events: Vec<StreamedEvent>) -> crate::Result<Vec<StreamDelivery>> {
        let start_time = std::time::Instant::now();
        let mut deliveries = Vec::new();

        for event in events {
            let delivery_start = std::time::Instant::now();
            let delivery_id = Uuid::new_v4();

            // Send the bytes produced by the stream's configured serialization format.
            let payload = event.serialized_data.clone();

            // Create Kinesis record
            let partition_key = event.partition_key.clone().unwrap_or_else(|| {
                // Use event ID as fallback partition key
                event.event.event_id.to_string()
            });

            let put_result = self
                .client
                .put_record()
                .stream_name(&self.stream_name)
                .data(aws_sdk_kinesis::primitives::Blob::new(payload))
                .partition_key(&partition_key)
                .send()
                .await;

            let delivery = match put_result {
                Ok(output) => {
                    let sequence_number = output.sequence_number();
                    let shard_id = output.shard_id();
                    tracing::debug!(
                        "Published event to Kinesis: sequence={}, shard={}",
                        sequence_number,
                        shard_id
                    );

                    StreamDelivery {
                        delivery_id,
                        stream_id: Uuid::new_v4(), // Would be passed from StreamManager
                        event_id: event.event.event_id,
                        success: true,
                        error_message: None,
                        attempted_at: Utc::now(),
                        duration_ms: Some(
                            u64::try_from(delivery_start.elapsed().as_millis()).unwrap_or(u64::MAX),
                        ),
                        attempt_number: 1,
                        partition: Some(partition_key),
                    }
                }
                Err(e) => {
                    let error_msg = format!("Failed to publish to Kinesis: {}", e);
                    tracing::error!("{}", error_msg);

                    StreamDelivery {
                        delivery_id,
                        stream_id: Uuid::new_v4(),
                        event_id: event.event.event_id,
                        success: false,
                        error_message: Some(error_msg),
                        attempted_at: Utc::now(),
                        duration_ms: Some(
                            u64::try_from(delivery_start.elapsed().as_millis()).unwrap_or(u64::MAX),
                        ),
                        attempt_number: 1,
                        partition: Some(partition_key),
                    }
                }
            };

            deliveries.push(delivery);
        }

        tracing::info!(
            "Sent {} events to Kinesis stream '{}' in {}ms",
            deliveries.len(),
            self.stream_name,
            start_time.elapsed().as_millis()
        );

        Ok(deliveries)
    }

    async fn health_check(&self) -> crate::Result<bool> {
        // Try to describe the stream to verify connectivity and access
        match self
            .client
            .describe_stream()
            .stream_name(&self.stream_name)
            .send()
            .await
        {
            Ok(output) => {
                if let Some(stream_description) = output.stream_description() {
                    let status = stream_description.stream_status();
                    tracing::debug!("Kinesis stream '{}' status: {:?}", self.stream_name, status);
                    // Consider stream healthy if it's ACTIVE or UPDATING
                    match status {
                        aws_sdk_kinesis::types::StreamStatus::Active
                        | aws_sdk_kinesis::types::StreamStatus::Updating => Ok(true),
                        _ => {
                            tracing::warn!(
                                "Kinesis stream '{}' is not in a healthy state: {:?}",
                                self.stream_name,
                                status
                            );
                            Ok(false)
                        }
                    }
                } else {
                    tracing::warn!(
                        "Kinesis stream '{}' description not available",
                        self.stream_name
                    );
                    Ok(false)
                }
            }
            Err(e) => {
                tracing::warn!(
                    "Kinesis health check failed for stream '{}': {}",
                    self.stream_name,
                    e
                );
                Ok(false)
            }
        }
    }

    async fn get_stats(&self) -> crate::Result<HashMap<String, serde_json::Value>> {
        let mut stats = HashMap::new();
        stats.insert(
            "type".to_string(),
            serde_json::Value::String("kinesis".to_string()),
        );
        stats.insert(
            "region".to_string(),
            serde_json::Value::String(self.region.clone()),
        );
        stats.insert(
            "stream_name".to_string(),
            serde_json::Value::String(self.stream_name.clone()),
        );
        stats.insert(
            "credentials_configured".to_string(),
            serde_json::Value::Bool(
                self.access_key_id.is_some() && self.secret_access_key.is_some(),
            ),
        );
        if let Some(ref access_key_id) = self.access_key_id {
            stats.insert(
                "access_key_id".to_string(),
                serde_json::Value::String(format!("{}***", mask_prefix(access_key_id))),
            );
        }
        stats.insert(
            "config".to_string(),
            serde_json::Value::Object(
                self.config
                    .iter()
                    .map(|(k, v)| (k.clone(), serde_json::Value::String(v.clone())))
                    .collect(),
            ),
        );
        stats.insert("feature_enabled".to_string(), serde_json::Value::Bool(true));
        Ok(stats)
    }

    async fn shutdown(&self) -> crate::Result<()> {
        // AWS SDK clients don't require explicit shutdown
        tracing::info!(
            "Kinesis processor shutdown for stream '{}'.",
            self.stream_name
        );
        Ok(())
    }
}

#[cfg(not(feature = "kinesis"))]
/// Uninhabited stand-in compiled when the `kinesis` feature is disabled.
///
/// [`KinesisProcessor::new`] always fails, so a value of this type can never exist and
/// streams for this backend are rejected when they are configured.
pub struct KinesisProcessor(std::convert::Infallible);

#[cfg(not(feature = "kinesis"))]
impl KinesisProcessor {
    /// Always fails: AWS Kinesis support requires the `kinesis` feature.
    pub async fn new(
        _region: String,
        _stream_name: String,
        _access_key_id: Option<String>,
        _secret_access_key: Option<String>,
        _config: HashMap<String, String>,
    ) -> crate::Result<Self> {
        Err(HammerworkError::Streaming {
            message: "AWS Kinesis support requires the `kinesis` feature of hammerwork".to_string(),
        })
    }
}

#[cfg(not(feature = "kinesis"))]
#[async_trait::async_trait]
impl StreamProcessor for KinesisProcessor {
    async fn send_batch(&self, _events: Vec<StreamedEvent>) -> crate::Result<Vec<StreamDelivery>> {
        match self.0 {}
    }

    async fn health_check(&self) -> crate::Result<bool> {
        match self.0 {}
    }

    async fn get_stats(&self) -> crate::Result<HashMap<String, serde_json::Value>> {
        match self.0 {}
    }

    async fn shutdown(&self) -> crate::Result<()> {
        match self.0 {}
    }
}

/// Google Cloud Pub/Sub processor for streaming job events
#[cfg(feature = "google-pubsub")]
pub struct PubSubProcessor {
    project_id: String,
    topic_name: String,
    service_account_key: Option<String>,
    config: HashMap<String, String>,
    topic: Topic,
    publisher: Publisher,
}

#[cfg(feature = "google-pubsub")]
impl PubSubProcessor {
    pub async fn new(
        project_id: String,
        topic_name: String,
        service_account_key: Option<String>,
        config: HashMap<String, String>,
    ) -> crate::Result<Self> {
        // Initialize Google Cloud Pub/Sub client
        let client_config = if let Some(service_account_json) = &service_account_key {
            // Use provided service account credentials
            let credentials = serde_json::from_str::<CredentialsFile>(service_account_json)
                .map_err(|e| HammerworkError::Streaming {
                    message: format!("Invalid service account credentials: {}", e),
                })?;

            PubSubClientConfig::default()
                .with_credentials(credentials)
                .await
                .map_err(|e| HammerworkError::Streaming {
                    message: format!("Failed to configure Pub/Sub client: {}", e),
                })?
        } else {
            // Use default credentials (Application Default Credentials)
            PubSubClientConfig::default()
                .with_auth()
                .await
                .map_err(|e| HammerworkError::Streaming {
                    message: format!(
                        "Failed to initialize Pub/Sub client with default credentials: {}",
                        e
                    ),
                })?
        };

        Self::with_client_config(
            project_id,
            topic_name,
            service_account_key,
            config,
            client_config,
        )
        .await
    }

    /// Build the processor from a ready Pub/Sub client configuration.
    async fn with_client_config(
        project_id: String,
        topic_name: String,
        service_account_key: Option<String>,
        config: HashMap<String, String>,
        client_config: PubSubClientConfig,
    ) -> crate::Result<Self> {
        // Validate the health check timeout up front rather than on the first check.
        parse_health_check_timeout_ms(&config)?;

        crate::ensure_rustls_crypto_provider();
        let client = Client::new(client_config)
            .await
            .map_err(|e| HammerworkError::Streaming {
                message: format!("Failed to create Pub/Sub client: {}", e),
            })?;

        let topic = client.topic(&topic_name);
        let publisher = topic.new_publisher(None);

        Ok(Self {
            project_id,
            topic_name,
            service_account_key,
            config,
            topic,
            publisher,
        })
    }
}

#[cfg(feature = "google-pubsub")]
#[async_trait::async_trait]
impl StreamProcessor for PubSubProcessor {
    async fn send_batch(&self, events: Vec<StreamedEvent>) -> crate::Result<Vec<StreamDelivery>> {
        let start_time = std::time::Instant::now();
        let mut deliveries = Vec::new();

        for event in events {
            let delivery_start = std::time::Instant::now();
            let delivery_id = Uuid::new_v4();

            // Send the bytes produced by the stream's configured serialization format.
            let payload = event.serialized_data.clone();

            // Create message attributes
            let mut attributes = std::collections::HashMap::new();
            attributes.insert("event_id".to_string(), event.event.event_id.to_string());
            attributes.insert("job_id".to_string(), event.event.job_id.to_string());
            attributes.insert("queue_name".to_string(), event.event.queue_name.clone());
            attributes.insert(
                "event_type".to_string(),
                format!("{:?}", event.event.event_type),
            );
            attributes.insert("timestamp".to_string(), event.event.timestamp.to_rfc3339());

            if let Some(partition_key) = &event.partition_key {
                attributes.insert("partition_key".to_string(), partition_key.clone());
            }

            // Create PubsubMessage
            let message = PubsubMessage {
                data: payload,
                attributes,
                message_id: String::new(),
                publish_time: None,
                ordering_key: event.partition_key.clone().unwrap_or_default(),
            };

            // Publish the message using the publisher
            let awaiter = self.publisher.publish(message).await;

            let delivery = match awaiter.get().await {
                Ok(message_id) => {
                    tracing::debug!("Published message to Pub/Sub: {}", message_id);
                    StreamDelivery {
                        delivery_id,
                        stream_id: Uuid::new_v4(), // Would be passed from StreamManager
                        event_id: event.event.event_id,
                        success: true,
                        error_message: None,
                        attempted_at: Utc::now(),
                        duration_ms: Some(
                            u64::try_from(delivery_start.elapsed().as_millis()).unwrap_or(u64::MAX),
                        ),
                        attempt_number: 1,
                        partition: event.partition_key,
                    }
                }
                Err(e) => {
                    let error_msg = format!("Failed to publish to Pub/Sub: {}", e);
                    tracing::error!("{}", error_msg);
                    StreamDelivery {
                        delivery_id,
                        stream_id: Uuid::new_v4(),
                        event_id: event.event.event_id,
                        success: false,
                        error_message: Some(error_msg),
                        attempted_at: Utc::now(),
                        duration_ms: Some(
                            u64::try_from(delivery_start.elapsed().as_millis()).unwrap_or(u64::MAX),
                        ),
                        attempt_number: 1,
                        partition: event.partition_key,
                    }
                }
            };

            deliveries.push(delivery);
        }

        tracing::info!(
            "Sent {} events to Pub/Sub topic '{}' in {}ms",
            deliveries.len(),
            self.topic_name,
            start_time.elapsed().as_millis()
        );

        Ok(deliveries)
    }

    /// Checks that the topic exists by fetching it from Pub/Sub.
    ///
    /// Returns `Ok(true)` when the topic exists, `Ok(false)` when it does not or
    /// Pub/Sub cannot be reached within `health.check.timeout.ms` (default 5000), and
    /// an error when Pub/Sub rejects the request (for example missing permissions or
    /// invalid credentials), since retrying will not fix that.
    async fn health_check(&self) -> crate::Result<bool> {
        use google_cloud_gax::grpc::Code;
        use google_cloud_gax::retry::RetrySetting;

        let timeout = Duration::from_millis(parse_health_check_timeout_ms(&self.config)?);
        // A single attempt: the timeout above bounds the whole check.
        let no_retry = RetrySetting {
            take: 0,
            codes: Vec::new(),
            ..Default::default()
        };

        match tokio::time::timeout(timeout, self.topic.exists(Some(no_retry))).await {
            Ok(Ok(true)) => {
                tracing::debug!(
                    "Pub/Sub health check passed for topic '{}'",
                    self.topic_name
                );
                Ok(true)
            }
            Ok(Ok(false)) => {
                tracing::warn!("Pub/Sub topic '{}' does not exist", self.topic_name);
                Ok(false)
            }
            Ok(Err(status)) => match status.code() {
                Code::Unavailable | Code::DeadlineExceeded | Code::Unknown | Code::Aborted => {
                    tracing::error!(
                        "Pub/Sub health check for topic '{}' failed: {}",
                        self.topic_name,
                        status
                    );
                    Ok(false)
                }
                _ => Err(HammerworkError::Streaming {
                    message: format!(
                        "Pub/Sub health check for topic '{}' was rejected: {}",
                        self.topic_name, status
                    ),
                }),
            },
            Err(_) => {
                tracing::error!(
                    "Pub/Sub health check for topic '{}' timed out after {:?}",
                    self.topic_name,
                    timeout
                );
                Ok(false)
            }
        }
    }

    async fn get_stats(&self) -> crate::Result<HashMap<String, serde_json::Value>> {
        let mut stats = HashMap::new();
        stats.insert(
            "type".to_string(),
            serde_json::Value::String("pubsub".to_string()),
        );
        stats.insert(
            "project_id".to_string(),
            serde_json::Value::String(self.project_id.clone()),
        );
        stats.insert(
            "topic_name".to_string(),
            serde_json::Value::String(self.topic_name.clone()),
        );
        stats.insert(
            "service_account_configured".to_string(),
            serde_json::Value::Bool(self.service_account_key.is_some()),
        );
        stats.insert(
            "config".to_string(),
            serde_json::Value::Object(
                self.config
                    .iter()
                    .map(|(k, v)| (k.clone(), serde_json::Value::String(v.clone())))
                    .collect(),
            ),
        );

        // Add publisher stats if available
        // Note: google-cloud-pubsub doesn't expose detailed publisher stats
        // This would be enhanced with custom metrics collection
        stats.insert("feature_enabled".to_string(), serde_json::Value::Bool(true));

        Ok(stats)
    }

    async fn shutdown(&self) -> crate::Result<()> {
        // The publisher will be dropped and cleaned up automatically
        tracing::info!(
            "Pub/Sub publisher shutdown for topic '{}'.",
            self.topic_name
        );
        Ok(())
    }
}

#[cfg(not(feature = "google-pubsub"))]
/// Uninhabited stand-in compiled when the `google-pubsub` feature is disabled.
///
/// [`PubSubProcessor::new`] always fails, so a value of this type can never exist and
/// streams for this backend are rejected when they are configured.
pub struct PubSubProcessor(std::convert::Infallible);

#[cfg(not(feature = "google-pubsub"))]
impl PubSubProcessor {
    /// Always fails: Google Pub/Sub support requires the `google-pubsub` feature.
    pub async fn new(
        _project_id: String,
        _topic_name: String,
        _service_account_key: Option<String>,
        _config: HashMap<String, String>,
    ) -> crate::Result<Self> {
        Err(HammerworkError::Streaming {
            message: "Google Pub/Sub support requires the `google-pubsub` feature of hammerwork"
                .to_string(),
        })
    }
}

#[cfg(not(feature = "google-pubsub"))]
#[async_trait::async_trait]
impl StreamProcessor for PubSubProcessor {
    async fn send_batch(&self, _events: Vec<StreamedEvent>) -> crate::Result<Vec<StreamDelivery>> {
        match self.0 {}
    }

    async fn health_check(&self) -> crate::Result<bool> {
        match self.0 {}
    }

    async fn get_stats(&self) -> crate::Result<HashMap<String, serde_json::Value>> {
        match self.0 {}
    }

    async fn shutdown(&self) -> crate::Result<()> {
        match self.0 {}
    }
}

/// Helper for creating stream configurations
impl StreamConfig {
    /// Create a new stream configuration
    pub fn new(name: String, backend: StreamBackend) -> Self {
        Self {
            id: Uuid::new_v4(),
            name,
            backend,
            filter: EventFilter::default(),
            partitioning: PartitioningStrategy::None,
            serialization: SerializationFormat::Json,
            retry_policy: StreamRetryPolicy::default(),
            enabled: true,
            buffer_config: BufferConfig::default(),
        }
    }

    /// Set event filter
    pub fn with_filter(mut self, filter: EventFilter) -> Self {
        self.filter = filter;
        self
    }

    /// Set partitioning strategy
    pub fn with_partitioning(mut self, partitioning: PartitioningStrategy) -> Self {
        self.partitioning = partitioning;
        self
    }

    /// Set serialization format
    pub fn with_serialization(mut self, serialization: SerializationFormat) -> Self {
        self.serialization = serialization;
        self
    }

    /// Set retry policy
    pub fn with_retry_policy(mut self, retry_policy: StreamRetryPolicy) -> Self {
        self.retry_policy = retry_policy;
        self
    }

    /// Set buffer configuration
    pub fn with_buffer_config(mut self, buffer_config: BufferConfig) -> Self {
        self.buffer_config = buffer_config;
        self
    }

    /// Enable or disable stream
    pub fn enabled(mut self, enabled: bool) -> Self {
        self.enabled = enabled;
        self
    }
}

/// Decides whether the `call`-th (0-based) `send_batch` call fails as a whole.
#[cfg(test)]
type BatchFailurePlan = Arc<dyn Fn(usize) -> bool + Send + Sync>;

/// Decides whether one event fails within the `call`-th (0-based) `send_batch` call.
#[cfg(test)]
type EventFailurePlan = Arc<dyn Fn(usize, &StreamedEvent) -> bool + Send + Sync>;

/// Broker-less processor for tests that need a working stream without any
/// backend feature enabled.
///
/// By default it acknowledges every event. It records every event it acknowledged
/// and how often each event was attempted, and can be scripted to fail whole
/// batches or single events. Clones share the recorded state, so a test can keep a
/// clone while the manager owns the processor.
#[cfg(test)]
#[derive(Clone, Default)]
struct InMemoryProcessor {
    /// Events acknowledged, in delivery order.
    delivered: Arc<std::sync::Mutex<Vec<StreamedEvent>>>,
    /// Number of attempts per event id.
    attempts: Arc<std::sync::Mutex<HashMap<Uuid, usize>>>,
    /// Number of `send_batch` calls so far.
    calls: Arc<std::sync::atomic::AtomicUsize>,
    /// Makes a whole `send_batch` call return an error.
    fail_batch: Option<BatchFailurePlan>,
    /// Makes single events come back with an unsuccessful delivery.
    fail_event: Option<EventFailurePlan>,
}

#[cfg(test)]
impl InMemoryProcessor {
    fn failing_batches(mut self, plan: impl Fn(usize) -> bool + Send + Sync + 'static) -> Self {
        self.fail_batch = Some(Arc::new(plan));
        self
    }

    fn failing_events(
        mut self,
        plan: impl Fn(usize, &StreamedEvent) -> bool + Send + Sync + 'static,
    ) -> Self {
        self.fail_event = Some(Arc::new(plan));
        self
    }

    fn delivered_ids(&self) -> Vec<Uuid> {
        self.delivered
            .lock()
            .unwrap()
            .iter()
            .map(|e| e.event.event_id)
            .collect()
    }

    fn attempts_for(&self, event_id: Uuid) -> usize {
        self.attempts
            .lock()
            .unwrap()
            .get(&event_id)
            .copied()
            .unwrap_or(0)
    }

    fn calls(&self) -> usize {
        self.calls.load(std::sync::atomic::Ordering::SeqCst)
    }
}

#[cfg(test)]
#[async_trait::async_trait]
impl StreamProcessor for InMemoryProcessor {
    async fn send_batch(&self, events: Vec<StreamedEvent>) -> crate::Result<Vec<StreamDelivery>> {
        let call = self.calls.fetch_add(1, std::sync::atomic::Ordering::SeqCst);
        {
            let mut attempts = self.attempts.lock().unwrap();
            for event in &events {
                *attempts.entry(event.event.event_id).or_default() += 1;
            }
        }
        if self.fail_batch.as_ref().is_some_and(|fail| fail(call)) {
            return Err(HammerworkError::Streaming {
                message: format!("in-memory backend rejected batch {call}"),
            });
        }

        let mut deliveries = Vec::with_capacity(events.len());
        for event in events {
            let failed = self
                .fail_event
                .as_ref()
                .is_some_and(|fail| fail(call, &event));
            deliveries.push(StreamDelivery {
                delivery_id: Uuid::new_v4(),
                stream_id: Uuid::nil(),
                event_id: event.event.event_id,
                success: !failed,
                error_message: failed
                    .then(|| format!("in-memory backend rejected event in call {call}")),
                attempted_at: Utc::now(),
                duration_ms: Some(0),
                attempt_number: 1,
                partition: event.partition_key.clone(),
            });
            if !failed {
                self.delivered.lock().unwrap().push(event);
            }
        }
        Ok(deliveries)
    }

    async fn health_check(&self) -> crate::Result<bool> {
        Ok(true)
    }

    async fn get_stats(&self) -> crate::Result<HashMap<String, serde_json::Value>> {
        Ok(HashMap::new())
    }

    async fn shutdown(&self) -> crate::Result<()> {
        Ok(())
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::{events::JobLifecycleEventType, priority::JobPriority};

    #[test]
    fn test_stream_backend_debug_hides_credentials() {
        let config = HashMap::from([("sasl.password".to_string(), "hunter2-sasl".to_string())]);
        for backend in [
            StreamBackend::Kafka {
                brokers: vec!["kafka:9092".to_string()],
                topic: "events".to_string(),
                config: config.clone(),
            },
            StreamBackend::Kinesis {
                region: "us-east-1".to_string(),
                stream_name: "events".to_string(),
                access_key_id: Some("AKIDEXAMPLE".to_string()),
                secret_access_key: Some("hunter2-aws".to_string()),
                config: config.clone(),
            },
            StreamBackend::PubSub {
                project_id: "project".to_string(),
                topic_name: "events".to_string(),
                service_account_key: Some("hunter2-gcp".to_string()),
                config: config.clone(),
            },
        ] {
            let debug = format!("{backend:?}");
            assert!(!debug.contains("hunter2"), "{debug}");
            assert!(debug.contains("sasl.password"), "{debug}");
        }
    }

    #[test]
    fn test_clamp_processor_permits() {
        assert_eq!(clamp_processor_permits(0), 1);
        assert_eq!(clamp_processor_permits(9), 9);
        assert_eq!(clamp_processor_permits(usize::MAX), Semaphore::MAX_PERMITS);
    }

    #[test]
    fn test_stream_manager_new_survives_extreme_concurrency() {
        for value in [0, usize::MAX] {
            let manager = StreamManager::new(
                Arc::new(EventManager::new_default()),
                StreamManagerConfig {
                    max_concurrent_processors: value,
                    ..Default::default()
                },
            );
            let permits = manager.processing_semaphore.available_permits();
            assert!((1..=Semaphore::MAX_PERMITS).contains(&permits));
        }
    }

    #[test]
    fn test_mask_prefix_is_char_boundary_safe() {
        assert_eq!(mask_prefix("AKIAIOSFODNN7"), "AKIA");
        assert_eq!(mask_prefix("ab"), "ab");
        // Four-byte prefix would split the second character.
        assert_eq!(mask_prefix("日本語キー"), "日本語キ");
    }

    #[test]
    fn test_parse_health_check_timeout_ms() {
        let mut config = HashMap::new();
        assert_eq!(parse_health_check_timeout_ms(&config).unwrap(), 5000);
        config.insert("health.check.timeout.ms".to_string(), "250".to_string());
        assert_eq!(parse_health_check_timeout_ms(&config).unwrap(), 250);
        config.insert("health.check.timeout.ms".to_string(), "abc".to_string());
        assert!(parse_health_check_timeout_ms(&config).is_err());
    }

    #[tokio::test]
    async fn test_stream_manager_from_config() {
        let mut config = crate::config::StreamingConfigs::default();
        config.global_settings.max_concurrent_processors = 3;
        config.global_settings.log_operations = false;
        config.global_settings.global_flush_interval_secs = 2;

        let manager_config = StreamManagerConfig::from(&config.global_settings);
        assert_eq!(manager_config.max_concurrent_processors, 3);
        assert!(!manager_config.log_operations);
        assert_eq!(manager_config.global_flush_interval_secs, 2);

        let manager = StreamManager::from_config(Arc::new(EventManager::new_default()), &config)
            .await
            .unwrap();
        assert!(manager.list_streams().await.is_empty());
    }

    #[test]
    fn test_stream_config_creation() {
        let backend = StreamBackend::Kafka {
            brokers: vec!["localhost:9092".to_string()],
            topic: "job-events".to_string(),
            config: HashMap::new(),
        };

        let stream = StreamConfig::new("test-stream".to_string(), backend)
            .with_partitioning(PartitioningStrategy::QueueName)
            .with_serialization(SerializationFormat::Json)
            .enabled(true);

        assert_eq!(stream.name, "test-stream");
        assert!(stream.enabled);
        assert!(matches!(
            stream.partitioning,
            PartitioningStrategy::QueueName
        ));
        assert!(matches!(stream.serialization, SerializationFormat::Json));
    }

    #[test]
    fn test_partitioning_strategies() {
        let event = JobLifecycleEvent {
            event_id: Uuid::new_v4(),
            job_id: Uuid::new_v4(),
            queue_name: "email_queue".to_string(),
            event_type: JobLifecycleEventType::Completed,
            priority: JobPriority::High,
            timestamp: Utc::now(),
            processing_time_ms: Some(1000),
            error: None,
            payload: None,
            metadata: {
                let mut metadata = HashMap::new();
                metadata.insert("source".to_string(), "api".to_string());
                metadata
            },
        };

        // Test different partitioning strategies
        assert_eq!(
            StreamManager::calculate_partition_key(&event, &PartitioningStrategy::None),
            None
        );

        assert_eq!(
            StreamManager::calculate_partition_key(&event, &PartitioningStrategy::QueueName),
            Some("email_queue".to_string())
        );

        assert_eq!(
            StreamManager::calculate_partition_key(&event, &PartitioningStrategy::Priority),
            Some("high".to_string())
        );

        assert_eq!(
            StreamManager::calculate_partition_key(&event, &PartitioningStrategy::EventType),
            Some("completed".to_string())
        );

        assert_eq!(
            StreamManager::calculate_partition_key(
                &event,
                &PartitioningStrategy::Custom {
                    metadata_key: "source".to_string()
                }
            ),
            Some("api".to_string())
        );

        // Test hash partitioning
        let hash_result = StreamManager::calculate_partition_key(
            &event,
            &PartitioningStrategy::Hash {
                fields: vec![PartitionField::QueueName, PartitionField::Priority],
            },
        );
        assert!(hash_result.is_some());
    }

    #[test]
    fn test_stream_backends() {
        let kafka = StreamBackend::Kafka {
            brokers: vec!["localhost:9092".to_string()],
            topic: "events".to_string(),
            config: HashMap::new(),
        };

        let kinesis = StreamBackend::Kinesis {
            region: "us-east-1".to_string(),
            stream_name: "job-events".to_string(),
            access_key_id: None,
            secret_access_key: None,
            config: HashMap::new(),
        };

        let pubsub = StreamBackend::PubSub {
            project_id: "my-project".to_string(),
            topic_name: "job-events".to_string(),
            service_account_key: None,
            config: HashMap::new(),
        };

        match kafka {
            StreamBackend::Kafka { brokers, topic, .. } => {
                assert_eq!(brokers.len(), 1);
                assert_eq!(topic, "events");
            }
            _ => panic!("Wrong backend type"),
        }

        match kinesis {
            StreamBackend::Kinesis {
                region,
                stream_name,
                ..
            } => {
                assert_eq!(region, "us-east-1");
                assert_eq!(stream_name, "job-events");
            }
            _ => panic!("Wrong backend type"),
        }

        match pubsub {
            StreamBackend::PubSub {
                project_id,
                topic_name,
                ..
            } => {
                assert_eq!(project_id, "my-project");
                assert_eq!(topic_name, "job-events");
            }
            _ => panic!("Wrong backend type"),
        }
    }

    #[test]
    fn test_retry_policy() {
        let policy = StreamRetryPolicy {
            max_attempts: 3,
            initial_delay_secs: 1,
            max_delay_secs: 60,
            backoff_multiplier: 2.0,
            use_jitter: false,
        };

        assert_eq!(policy.max_attempts, 3);
        assert_eq!(policy.initial_delay_secs, 1);
        assert_eq!(policy.backoff_multiplier, 2.0);
        assert!(!policy.use_jitter);
    }

    #[test]
    fn test_buffer_config() {
        let buffer = BufferConfig {
            max_events: 500,
            max_buffer_time_secs: 10,
            batch_size: 50,
        };

        assert_eq!(buffer.max_events, 500);
        assert_eq!(buffer.max_buffer_time_secs, 10);
        assert_eq!(buffer.batch_size, 50);
    }

    #[test]
    fn test_serialization_formats() {
        let json = SerializationFormat::Json;
        let avro = SerializationFormat::Avro {
            schema_registry_url: "http://localhost:8081".to_string(),
        };
        let protobuf = SerializationFormat::Protobuf {
            schema_definition: "syntax = \"proto3\";".to_string(),
        };
        let msgpack = SerializationFormat::MessagePack;

        assert!(matches!(json, SerializationFormat::Json));
        assert!(matches!(avro, SerializationFormat::Avro { .. }));
        assert!(matches!(protobuf, SerializationFormat::Protobuf { .. }));
        assert!(matches!(msgpack, SerializationFormat::MessagePack));
    }

    fn create_test_job_lifecycle_event() -> JobLifecycleEvent {
        JobLifecycleEvent {
            event_id: Uuid::new_v4(),
            job_id: Uuid::new_v4(),
            queue_name: "test_queue".to_string(),
            event_type: JobLifecycleEventType::Started,
            priority: crate::priority::JobPriority::Normal,
            timestamp: chrono::Utc::now(),
            processing_time_ms: Some(150),
            error: None,
            payload: Some(serde_json::json!({"message": "test payload"})),
            metadata: {
                let mut metadata = HashMap::new();
                metadata.insert("test_key".to_string(), "test_value".to_string());
                metadata.insert("priority".to_string(), "normal".to_string());
                metadata
            },
        }
    }

    #[test]
    fn test_json_serialization() {
        let event = create_test_job_lifecycle_event();
        let stream_config = StreamConfig {
            name: "test_stream".to_string(),
            serialization: SerializationFormat::Json,
            ..Default::default()
        };

        let result = StreamManager::prepare_streamed_event(event.clone(), &stream_config);
        assert!(result.is_ok());

        let streamed_event = result.unwrap();
        assert!(!streamed_event.serialized_data.is_empty());

        // Verify we can deserialize back to the original event
        let deserialized: JobLifecycleEvent =
            serde_json::from_slice(&streamed_event.serialized_data).unwrap();
        assert_eq!(deserialized.queue_name, event.queue_name);
        assert_eq!(deserialized.event_type, event.event_type);
        assert_eq!(deserialized.priority, event.priority);
        assert_eq!(deserialized.metadata, event.metadata);
    }

    #[test]
    #[cfg(feature = "streaming")]
    fn test_messagepack_serialization() {
        let event = create_test_job_lifecycle_event();
        let stream_config = StreamConfig {
            name: "test_stream".to_string(),
            serialization: SerializationFormat::MessagePack,
            ..Default::default()
        };

        let result = StreamManager::prepare_streamed_event(event.clone(), &stream_config);
        assert!(result.is_ok());

        let streamed_event = result.unwrap();
        assert!(!streamed_event.serialized_data.is_empty());

        // Verify we can deserialize back to the original event
        let deserialized: JobLifecycleEvent =
            rmp_serde::from_slice(&streamed_event.serialized_data).unwrap();
        assert_eq!(deserialized.queue_name, event.queue_name);
        assert_eq!(deserialized.event_type, event.event_type);
        assert_eq!(deserialized.priority, event.priority);
        assert_eq!(deserialized.metadata, event.metadata);

        // MessagePack should be more compact than JSON for this data
        let json_size = serde_json::to_vec(&event).unwrap().len();
        assert!(streamed_event.serialized_data.len() <= json_size);
    }

    #[test]
    #[cfg(not(feature = "streaming"))]
    fn test_messagepack_serialization_requires_feature() {
        let event = create_test_job_lifecycle_event();
        let stream_config = StreamConfig {
            name: "test_stream".to_string(),
            serialization: SerializationFormat::MessagePack,
            ..Default::default()
        };

        let result = StreamManager::prepare_streamed_event(event, &stream_config);
        assert!(result.is_err());

        let error = result.unwrap_err();
        if let crate::error::HammerworkError::Streaming { message } = error {
            assert!(message.contains("streaming"));
            assert!(message.contains("feature"));
        } else {
            panic!("Expected Streaming error");
        }
    }

    #[test]
    #[cfg(feature = "streaming")]
    fn test_avro_serialization() {
        let event = create_test_job_lifecycle_event();
        let stream_config = StreamConfig {
            name: "test_stream".to_string(),
            serialization: SerializationFormat::Avro {
                schema_registry_url: "http://localhost:8081".to_string(),
            },
            ..Default::default()
        };

        let result = StreamManager::prepare_streamed_event(event.clone(), &stream_config);
        assert!(result.is_ok());

        let streamed_event = result.unwrap();
        assert!(!streamed_event.serialized_data.is_empty());

        // Verify the Avro data has the expected binary format header
        // Avro binary format starts with a magic byte sequence
        assert!(streamed_event.serialized_data.len() > 10); // Should have header + data
    }

    #[test]
    #[cfg(not(feature = "streaming"))]
    fn test_avro_serialization_requires_feature() {
        let event = create_test_job_lifecycle_event();
        let stream_config = StreamConfig {
            name: "test_stream".to_string(),
            serialization: SerializationFormat::Avro {
                schema_registry_url: "http://localhost:8081".to_string(),
            },
            ..Default::default()
        };

        let result = StreamManager::prepare_streamed_event(event, &stream_config);
        assert!(result.is_err());

        let error = result.unwrap_err();
        if let crate::error::HammerworkError::Streaming { message } = error {
            assert!(message.contains("streaming"));
            assert!(message.contains("feature"));
        } else {
            panic!("Expected Streaming error");
        }
    }

    #[test]
    #[cfg(feature = "streaming")]
    fn test_protobuf_serialization() {
        let event = create_test_job_lifecycle_event();
        let stream_config = StreamConfig {
            name: "test_stream".to_string(),
            serialization: SerializationFormat::Protobuf {
                schema_definition: "syntax = \"proto3\";".to_string(),
            },
            ..Default::default()
        };

        let result = StreamManager::prepare_streamed_event(event.clone(), &stream_config);
        assert!(result.is_ok());

        let streamed_event = result.unwrap();
        assert!(!streamed_event.serialized_data.is_empty());

        // Verify the Protobuf data has a reasonable binary format
        // Protobuf is typically very compact
        assert!(streamed_event.serialized_data.len() > 5); // Should have meaningful data

        // Protobuf should be more compact than JSON for structured data
        let json_size = serde_json::to_vec(&event).unwrap().len();
        assert!(streamed_event.serialized_data.len() <= json_size);
    }

    #[test]
    #[cfg(not(feature = "streaming"))]
    fn test_protobuf_serialization_requires_feature() {
        let event = create_test_job_lifecycle_event();
        let stream_config = StreamConfig {
            name: "test_stream".to_string(),
            serialization: SerializationFormat::Protobuf {
                schema_definition: "syntax = \"proto3\";".to_string(),
            },
            ..Default::default()
        };

        let result = StreamManager::prepare_streamed_event(event, &stream_config);
        assert!(result.is_err());

        let error = result.unwrap_err();
        if let crate::error::HammerworkError::Streaming { message } = error {
            assert!(message.contains("streaming"));
            assert!(message.contains("feature"));
        } else {
            panic!("Expected Streaming error");
        }
    }

    #[test]
    #[cfg(feature = "streaming")]
    fn test_serialization_format_size_comparison() {
        let event = create_test_job_lifecycle_event();

        // Test all serialization formats
        let json_config = StreamConfig {
            name: "json_stream".to_string(),
            serialization: SerializationFormat::Json,
            ..Default::default()
        };

        let msgpack_config = StreamConfig {
            name: "msgpack_stream".to_string(),
            serialization: SerializationFormat::MessagePack,
            ..Default::default()
        };

        let avro_config = StreamConfig {
            name: "avro_stream".to_string(),
            serialization: SerializationFormat::Avro {
                schema_registry_url: "http://localhost:8081".to_string(),
            },
            ..Default::default()
        };

        let protobuf_config = StreamConfig {
            name: "protobuf_stream".to_string(),
            serialization: SerializationFormat::Protobuf {
                schema_definition: "syntax = \"proto3\";".to_string(),
            },
            ..Default::default()
        };

        let json_result =
            StreamManager::prepare_streamed_event(event.clone(), &json_config).unwrap();
        let msgpack_result =
            StreamManager::prepare_streamed_event(event.clone(), &msgpack_config).unwrap();
        let avro_result =
            StreamManager::prepare_streamed_event(event.clone(), &avro_config).unwrap();
        let protobuf_result =
            StreamManager::prepare_streamed_event(event.clone(), &protobuf_config).unwrap();

        // All formats should produce non-empty data
        assert!(!json_result.serialized_data.is_empty());
        assert!(!msgpack_result.serialized_data.is_empty());
        assert!(!avro_result.serialized_data.is_empty());
        assert!(!protobuf_result.serialized_data.is_empty());

        // Binary formats are typically more compact than JSON
        let json_size = json_result.serialized_data.len();
        let msgpack_size = msgpack_result.serialized_data.len();
        let protobuf_size = protobuf_result.serialized_data.len();

        // MessagePack and Protobuf should generally be more compact than JSON
        assert!(
            msgpack_size <= json_size,
            "MessagePack ({} bytes) should be <= JSON ({} bytes)",
            msgpack_size,
            json_size
        );
        assert!(
            protobuf_size <= json_size,
            "Protobuf ({} bytes) should be <= JSON ({} bytes)",
            protobuf_size,
            json_size
        );
    }

    #[tokio::test]
    async fn test_stream_manager_creation() {
        let config = StreamManagerConfig::default();
        let event_manager = Arc::new(EventManager::new_default());
        let manager = StreamManager::new(event_manager, config);

        let stats = manager.get_stats().await;
        assert_eq!(stats.total_streams, 0);
        assert_eq!(stats.active_streams, 0);
    }

    #[tokio::test]
    async fn test_stream_manager_add_remove_stream() {
        let config = StreamManagerConfig::default();
        let event_manager = Arc::new(EventManager::new_default());
        let manager = StreamManager::new(event_manager, config);

        let backend = StreamBackend::Kafka {
            brokers: vec!["localhost:9092".to_string()],
            topic: "test-events".to_string(),
            config: HashMap::new(),
        };

        let stream_config = StreamConfig::new("test-stream".to_string(), backend)
            .with_partitioning(PartitioningStrategy::QueueName)
            .with_serialization(SerializationFormat::Json);

        let stream_id = stream_config.id;

        // Add stream
        manager.add_stream_for_test(stream_config).await.unwrap();

        let stats = manager.get_stats().await;
        assert_eq!(stats.total_streams, 1);
        assert_eq!(stats.active_streams, 1);

        // Remove stream
        manager.remove_stream(stream_id).await.unwrap();

        let stats = manager.get_stats().await;
        assert_eq!(stats.total_streams, 0);
        assert_eq!(stats.active_streams, 0);
    }

    #[tokio::test]
    async fn test_stream_manager_enable_disable() {
        let config = StreamManagerConfig::default();
        let event_manager = Arc::new(EventManager::new_default());
        let manager = StreamManager::new(event_manager, config);

        let backend = StreamBackend::Kinesis {
            region: "us-west-2".to_string(),
            stream_name: "test-stream".to_string(),
            access_key_id: None,
            secret_access_key: None,
            config: HashMap::new(),
        };

        let stream_config = StreamConfig::new("test-stream".to_string(), backend);
        let stream_id = stream_config.id;

        manager.add_stream_for_test(stream_config).await.unwrap();

        // Initially enabled
        let stats = manager.get_stats().await;
        assert_eq!(stats.active_streams, 1);

        // Disable stream
        manager.disable_stream(stream_id).await.unwrap();
        let stats = manager.get_stats().await;
        assert_eq!(stats.active_streams, 0);
        assert_eq!(stats.total_streams, 1); // Still exists but disabled

        // Re-enable stream
        manager.enable_stream(stream_id).await.unwrap();
        let stats = manager.get_stats().await;
        assert_eq!(stats.active_streams, 1);
    }

    #[test]
    fn test_streamed_event_creation() {
        let event = JobLifecycleEvent {
            event_id: Uuid::new_v4(),
            job_id: Uuid::new_v4(),
            queue_name: "test_queue".to_string(),
            event_type: JobLifecycleEventType::Completed,
            priority: JobPriority::Normal,
            timestamp: Utc::now(),
            processing_time_ms: Some(1500),
            error: None,
            payload: Some(serde_json::json!({"key": "value"})),
            metadata: HashMap::new(),
        };

        let _stream_config = StreamConfig::new(
            "test-stream".to_string(),
            StreamBackend::Kafka {
                brokers: vec!["localhost:9092".to_string()],
                topic: "events".to_string(),
                config: HashMap::new(),
            },
        );

        let streamed_event = StreamedEvent {
            event,
            partition_key: Some("partition-1".to_string()),
            serialized_data: b"serialized data".to_vec(),
            streamed_at: Utc::now(),
            headers: HashMap::new(),
        };

        assert!(streamed_event.partition_key.is_some());
        assert!(!streamed_event.serialized_data.is_empty());
        assert_eq!(
            streamed_event.event.event_type,
            JobLifecycleEventType::Completed
        );
    }

    #[test]
    fn test_stream_delivery_result() {
        let delivery = StreamDelivery {
            delivery_id: Uuid::new_v4(),
            stream_id: Uuid::new_v4(),
            event_id: Uuid::new_v4(),
            success: true,
            error_message: None,
            attempted_at: Utc::now(),
            duration_ms: Some(100),
            attempt_number: 1,
            partition: Some("partition-0".to_string()),
        };

        assert!(delivery.success);
        assert!(delivery.error_message.is_none());
        assert_eq!(delivery.attempt_number, 1);
        assert_eq!(delivery.duration_ms, Some(100));
    }

    #[test]
    fn test_stream_stats_aggregation() {
        let mut stats = StreamStats {
            stream_id: Uuid::new_v4(),
            total_events: 1000,
            successful_deliveries: 950,
            failed_deliveries: 50,
            success_rate: 0.95,
            avg_delivery_time_ms: 125.5,
            buffered_events: 25,
            last_success_at: Some(Utc::now() - chrono::Duration::minutes(2)),
            last_failure_at: Some(Utc::now() - chrono::Duration::minutes(10)),
            last_error: None,
            dropped_events: 0,
            calculated_at: Utc::now(),
        };

        // Verify stats calculations
        assert_eq!(
            stats.total_events,
            stats.successful_deliveries + stats.failed_deliveries
        );
        assert!((stats.success_rate - 0.95).abs() < f64::EPSILON);
        assert!(stats.last_success_at.is_some());
        assert!(stats.last_failure_at.is_some());
        assert_eq!(stats.buffered_events, 25);

        // Update stats with new successful delivery
        stats.total_events += 1;
        stats.successful_deliveries += 1;
        stats.success_rate = stats.successful_deliveries as f64 / stats.total_events as f64;

        assert_eq!(stats.total_events, 1001);
        assert_eq!(stats.successful_deliveries, 951);
        assert!((stats.success_rate - 951.0 / 1001.0).abs() < f64::EPSILON);
    }

    /// Kafka brokers for tests that need a real cluster, from `KAFKA_BROKERS`
    /// (comma-separated). These tests are `#[ignore]`d and are skipped when
    /// `KAFKA_BROKERS` is unset, so `--include-ignored` runs without a broker
    /// still pass.
    #[cfg(feature = "kafka")]
    fn test_kafka_brokers() -> Option<Vec<String>> {
        match std::env::var("KAFKA_BROKERS") {
            Ok(brokers) => Some(brokers.split(',').map(|b| b.trim().to_string()).collect()),
            Err(_) => {
                eprintln!("skipping: KAFKA_BROKERS is not set");
                None
            }
        }
    }

    /// Whether a test that creates a real Pub/Sub client can run. With the
    /// `google-pubsub` feature these tests are `#[ignore]`d and are skipped
    /// when `GOOGLE_APPLICATION_CREDENTIALS` is unset.
    #[cfg(feature = "google-pubsub")]
    fn pubsub_credentials_available() -> bool {
        if std::env::var_os("GOOGLE_APPLICATION_CREDENTIALS").is_none() {
            eprintln!("skipping: GOOGLE_APPLICATION_CREDENTIALS is not set");
            return false;
        }
        true
    }

    /// A Kafka processor pointed at a port nothing listens on, with short
    /// timeouts, to exercise failure handling without a broker.
    #[cfg(feature = "kafka")]
    async fn unreachable_kafka_processor() -> KafkaProcessor {
        let mut config = HashMap::new();
        config.insert("message.timeout.ms".to_string(), "500".to_string());
        config.insert("health.check.timeout.ms".to_string(), "500".to_string());
        KafkaProcessor::new(
            vec!["127.0.0.1:1".to_string()],
            "test-topic".to_string(),
            config,
        )
        .await
        .unwrap()
    }

    #[cfg(feature = "kafka")]
    #[tokio::test]
    async fn test_kafka_health_check_unreachable_broker() {
        let processor = unreachable_kafka_processor().await;
        assert!(!processor.health_check().await.unwrap());
    }

    #[cfg(feature = "kafka")]
    #[tokio::test]
    async fn test_kafka_batch_sending_unreachable_broker() {
        let processor = unreachable_kafka_processor().await;

        let event = JobLifecycleEvent {
            event_id: Uuid::new_v4(),
            job_id: Uuid::new_v4(),
            queue_name: "test".to_string(),
            event_type: JobLifecycleEventType::Completed,
            priority: JobPriority::Normal,
            timestamp: Utc::now(),
            processing_time_ms: Some(1000),
            error: None,
            payload: None,
            metadata: HashMap::new(),
        };
        let event_id = event.event_id;

        let deliveries = processor
            .send_batch(vec![StreamedEvent {
                event,
                partition_key: Some("partition-0".to_string()),
                serialized_data: b"test data".to_vec(),
                streamed_at: Utc::now(),
                headers: HashMap::new(),
            }])
            .await
            .unwrap();

        assert_eq!(deliveries.len(), 1);
        assert_eq!(deliveries[0].event_id, event_id);
        assert!(!deliveries[0].success);
        assert!(deliveries[0].error_message.is_some());
    }

    #[cfg(feature = "google-pubsub")]
    #[tokio::test]
    async fn test_pubsub_processor_rejects_invalid_service_account_key() {
        let result = PubSubProcessor::new(
            "my-project".to_string(),
            "test-topic".to_string(),
            Some("not a service account key".to_string()),
            HashMap::new(),
        )
        .await;

        match result {
            Err(HammerworkError::Streaming { message }) => {
                assert!(message.contains("Invalid service account credentials"));
            }
            Err(e) => panic!("unexpected error: {e}"),
            Ok(_) => panic!("invalid service account key should be rejected"),
        }
    }

    #[cfg(feature = "kafka")]
    #[tokio::test]
    async fn test_kafka_processor_creation() {
        let processor = KafkaProcessor::new(
            vec!["localhost:9092".to_string()],
            "test-topic".to_string(),
            HashMap::new(),
        )
        .await
        .unwrap();

        assert_eq!(processor.topic, "test-topic");
    }

    #[cfg(feature = "kinesis")]
    #[tokio::test]
    async fn test_kinesis_processor_creation() {
        let processor = KinesisProcessor::new(
            "us-east-1".to_string(),
            "test-stream".to_string(),
            None,
            None,
            HashMap::new(),
        )
        .await
        .unwrap();

        assert_eq!(processor.region, "us-east-1");
        assert_eq!(processor.stream_name, "test-stream");
        assert!(processor.access_key_id.is_none());
    }

    #[cfg(feature = "google-pubsub")]
    #[tokio::test]
    #[ignore = "requires Google Pub/Sub: GOOGLE_APPLICATION_CREDENTIALS"]
    async fn test_pubsub_processor_creation() {
        if !pubsub_credentials_available() {
            return;
        }
        let processor = PubSubProcessor::new(
            "my-project".to_string(),
            "test-topic".to_string(),
            None,
            HashMap::new(),
        )
        .await
        .unwrap();

        assert_eq!(processor.project_id, "my-project");
        assert_eq!(processor.topic_name, "test-topic");
        assert!(processor.service_account_key.is_none());
    }

    #[cfg(feature = "kafka")]
    #[tokio::test]
    #[ignore = "requires Kafka: KAFKA_BROKERS"]
    async fn test_stream_processor_health_check() {
        let Some(brokers) = test_kafka_brokers() else {
            return;
        };
        let processor = KafkaProcessor::new(brokers, "test-topic".to_string(), HashMap::new())
            .await
            .unwrap();

        let health = processor.health_check().await.unwrap();
        assert!(health);
    }

    #[cfg(feature = "kafka")]
    #[tokio::test]
    #[ignore = "requires Kafka: KAFKA_BROKERS"]
    async fn test_stream_processor_batch_sending() {
        let Some(brokers) = test_kafka_brokers() else {
            return;
        };
        let processor = KafkaProcessor::new(brokers, "test-topic".to_string(), HashMap::new())
            .await
            .unwrap();

        let event = JobLifecycleEvent {
            event_id: Uuid::new_v4(),
            job_id: Uuid::new_v4(),
            queue_name: "test".to_string(),
            event_type: JobLifecycleEventType::Completed,
            priority: JobPriority::Normal,
            timestamp: Utc::now(),
            processing_time_ms: Some(1000),
            error: None,
            payload: None,
            metadata: HashMap::new(),
        };

        let streamed_event = StreamedEvent {
            event,
            partition_key: Some("partition-0".to_string()),
            serialized_data: b"test data".to_vec(),
            streamed_at: Utc::now(),
            headers: HashMap::new(),
        };

        let events = vec![streamed_event];
        let deliveries = processor.send_batch(events).await.unwrap();

        assert_eq!(deliveries.len(), 1);
        assert!(deliveries[0].success);
    }

    #[test]
    fn test_partition_field_combinations() {
        let event = JobLifecycleEvent {
            event_id: Uuid::new_v4(),
            job_id: Uuid::new_v4(),
            queue_name: "test_queue".to_string(),
            event_type: JobLifecycleEventType::Started,
            priority: JobPriority::High,
            timestamp: Utc::now(),
            processing_time_ms: None,
            error: None,
            payload: None,
            metadata: HashMap::new(),
        };

        // Test single field partitioning
        let queue_partition =
            StreamManager::calculate_partition_key(&event, &PartitioningStrategy::QueueName);
        assert_eq!(queue_partition, Some("test_queue".to_string()));

        let priority_partition =
            StreamManager::calculate_partition_key(&event, &PartitioningStrategy::Priority);
        assert_eq!(priority_partition, Some("high".to_string()));

        let event_type_partition =
            StreamManager::calculate_partition_key(&event, &PartitioningStrategy::EventType);
        assert_eq!(event_type_partition, Some("started".to_string()));

        // Test hash partitioning with multiple fields
        let hash_partition = StreamManager::calculate_partition_key(
            &event,
            &PartitioningStrategy::Hash {
                fields: vec![
                    PartitionField::QueueName,
                    PartitionField::Priority,
                    PartitionField::EventType,
                ],
            },
        );
        assert!(hash_partition.is_some());

        // Hash should be consistent for same input
        let hash_partition2 = StreamManager::calculate_partition_key(
            &event,
            &PartitioningStrategy::Hash {
                fields: vec![
                    PartitionField::QueueName,
                    PartitionField::Priority,
                    PartitionField::EventType,
                ],
            },
        );
        assert_eq!(hash_partition, hash_partition2);
    }

    #[test]
    fn test_stream_config_builder() {
        let backend = StreamBackend::PubSub {
            project_id: "test-project".to_string(),
            topic_name: "events".to_string(),
            service_account_key: None,
            config: HashMap::new(),
        };

        let filter = EventFilter::new().with_event_types(vec![
            JobLifecycleEventType::Completed,
            JobLifecycleEventType::Failed,
        ]);

        let retry_policy = StreamRetryPolicy {
            max_attempts: 5,
            initial_delay_secs: 2,
            max_delay_secs: 120,
            backoff_multiplier: 2.0,
            use_jitter: true,
        };

        let buffer_config = BufferConfig {
            max_events: 1000,
            max_buffer_time_secs: 30,
            batch_size: 100,
        };

        let stream = StreamConfig::new("comprehensive-stream".to_string(), backend)
            .with_filter(filter)
            .with_partitioning(PartitioningStrategy::Hash {
                fields: vec![PartitionField::QueueName, PartitionField::Priority],
            })
            .with_serialization(SerializationFormat::Avro {
                schema_registry_url: "http://schema-registry:8081".to_string(),
            })
            .with_retry_policy(retry_policy)
            .with_buffer_config(buffer_config)
            .enabled(true);

        assert_eq!(stream.name, "comprehensive-stream");
        assert!(stream.enabled);
        assert!(matches!(
            stream.partitioning,
            PartitioningStrategy::Hash { .. }
        ));
        assert!(matches!(
            stream.serialization,
            SerializationFormat::Avro { .. }
        ));
        assert_eq!(stream.retry_policy.max_attempts, 5);
        assert!(stream.retry_policy.use_jitter);
        assert_eq!(stream.buffer_config.max_events, 1000);
        assert_eq!(stream.filter.event_types.len(), 2);
    }

    #[test]
    fn test_stream_manager_config_defaults() {
        let config = StreamManagerConfig::default();

        assert_eq!(config.max_concurrent_processors, 50);
        assert!(config.log_operations);
        assert_eq!(config.global_flush_interval_secs, 10);
    }

    #[test]
    fn test_stream_config_serialization() {
        let backend = StreamBackend::Kinesis {
            region: "eu-west-1".to_string(),
            stream_name: "production-events".to_string(),
            access_key_id: Some("AKIA...".to_string()),
            secret_access_key: Some("secret".to_string()),
            config: {
                let mut config = HashMap::new();
                config.insert("batch_size".to_string(), "500".to_string());
                config
            },
        };

        let stream_config = StreamConfig::new("prod-stream".to_string(), backend)
            .with_partitioning(PartitioningStrategy::Custom {
                metadata_key: "tenant_id".to_string(),
            })
            .with_serialization(SerializationFormat::Protobuf {
                schema_definition: "syntax = \"proto3\"; message Event { string id = 1; }"
                    .to_string(),
            });

        let serialized = serde_json::to_string(&stream_config).unwrap();
        let deserialized: StreamConfig = serde_json::from_str(&serialized).unwrap();

        assert_eq!(deserialized.name, stream_config.name);
        assert_eq!(deserialized.enabled, stream_config.enabled);

        match &deserialized.backend {
            StreamBackend::Kinesis {
                region,
                stream_name,
                ..
            } => {
                assert_eq!(region, "eu-west-1");
                assert_eq!(stream_name, "production-events");
            }
            _ => panic!("Wrong backend type after deserialization"),
        }
    }

    #[cfg(feature = "google-pubsub")]
    #[tokio::test]
    #[ignore = "requires Google Pub/Sub: GOOGLE_APPLICATION_CREDENTIALS"]
    async fn test_stream_processor_stats() {
        if !pubsub_credentials_available() {
            return;
        }
        let processor = PubSubProcessor::new(
            "test-project".to_string(),
            "test-topic".to_string(),
            None,
            HashMap::new(),
        )
        .await
        .unwrap();

        let stats = processor.get_stats().await.unwrap();

        assert_eq!(
            stats["type"],
            serde_json::Value::String("pubsub".to_string())
        );
        assert_eq!(
            stats["project_id"],
            serde_json::Value::String("test-project".to_string())
        );
        assert_eq!(
            stats["topic_name"],
            serde_json::Value::String("test-topic".to_string())
        );
    }

    #[test]
    fn test_buffer_config_defaults() {
        let buffer = BufferConfig::default();

        assert_eq!(buffer.max_events, 1000);
        assert_eq!(buffer.max_buffer_time_secs, 5);
        assert_eq!(buffer.batch_size, 100);
    }

    #[test]
    fn test_stream_retry_policy_defaults() {
        let policy = StreamRetryPolicy::default();

        assert_eq!(policy.max_attempts, 5);
        assert_eq!(policy.initial_delay_secs, 1);
        assert_eq!(policy.max_delay_secs, 300);
        assert_eq!(policy.backoff_multiplier, 2.0);
        assert!(policy.use_jitter);
    }

    #[test]
    fn test_custom_metadata_partitioning() {
        let mut metadata = HashMap::new();
        metadata.insert("tenant_id".to_string(), "tenant_123".to_string());
        metadata.insert("region".to_string(), "us-west-2".to_string());

        let event = JobLifecycleEvent {
            event_id: Uuid::new_v4(),
            job_id: Uuid::new_v4(),
            queue_name: "multi_tenant_queue".to_string(),
            event_type: JobLifecycleEventType::Enqueued,
            priority: JobPriority::Normal,
            timestamp: Utc::now(),
            processing_time_ms: None,
            error: None,
            payload: None,
            metadata,
        };

        // Test custom metadata partitioning
        let tenant_partition = StreamManager::calculate_partition_key(
            &event,
            &PartitioningStrategy::Custom {
                metadata_key: "tenant_id".to_string(),
            },
        );
        assert_eq!(tenant_partition, Some("tenant_123".to_string()));

        let region_partition = StreamManager::calculate_partition_key(
            &event,
            &PartitioningStrategy::Custom {
                metadata_key: "region".to_string(),
            },
        );
        assert_eq!(region_partition, Some("us-west-2".to_string()));

        // Test non-existent metadata key
        let missing_partition = StreamManager::calculate_partition_key(
            &event,
            &PartitioningStrategy::Custom {
                metadata_key: "missing_key".to_string(),
            },
        );
        assert_eq!(missing_partition, None);
    }

    #[cfg(feature = "kafka")]
    #[tokio::test]
    async fn test_kafka_processor_configuration_usage() {
        let mut kafka_config = HashMap::new();
        kafka_config.insert("health.check.timeout.ms".to_string(), "1000".to_string());
        kafka_config.insert("linger.ms".to_string(), "50".to_string());

        let kafka_processor = KafkaProcessor::new(
            vec!["localhost:9092".to_string()],
            "test-topic".to_string(),
            kafka_config,
        )
        .await
        .unwrap();

        // Test that stats include configuration
        let stats = kafka_processor.get_stats().await.unwrap();
        assert_eq!(
            stats.get("type").unwrap(),
            &serde_json::Value::String("kafka".to_string())
        );
        assert!(stats.contains_key("config"));
    }

    #[cfg(feature = "kinesis")]
    #[tokio::test]
    async fn test_kinesis_processor_configuration_usage() {
        let mut kinesis_config = HashMap::new();
        kinesis_config.insert("retries".to_string(), "3".to_string());

        let kinesis_processor = KinesisProcessor::new(
            "us-west-2".to_string(),
            "test-stream".to_string(),
            Some("AKIA***".to_string()),
            Some("secret".to_string()),
            kinesis_config,
        )
        .await
        .unwrap();

        let kinesis_stats = kinesis_processor.get_stats().await.unwrap();
        assert_eq!(
            kinesis_stats.get("type").unwrap(),
            &serde_json::Value::String("kinesis".to_string())
        );
        assert!(kinesis_stats.contains_key("access_key_id"));
        assert!(kinesis_stats.contains_key("config"));
    }

    #[cfg(feature = "google-pubsub")]
    #[tokio::test]
    #[ignore = "requires Google Pub/Sub: GOOGLE_APPLICATION_CREDENTIALS"]
    async fn test_pubsub_processor_configuration_usage() {
        if !pubsub_credentials_available() {
            return;
        }
        // Test PubSub processor with service account
        let mut pubsub_config = HashMap::new();
        pubsub_config.insert("max_messages".to_string(), "1000".to_string());

        // The real client needs a valid service account key.
        let service_account_key = std::fs::read_to_string(
            std::env::var("GOOGLE_APPLICATION_CREDENTIALS")
                .expect("GOOGLE_APPLICATION_CREDENTIALS must point at a service account key"),
        )
        .expect("failed to read service account key");

        let pubsub_processor = PubSubProcessor::new(
            "my-project".to_string(),
            "test-topic".to_string(),
            Some(service_account_key),
            pubsub_config,
        )
        .await
        .unwrap();

        let pubsub_stats = pubsub_processor.get_stats().await.unwrap();
        assert_eq!(
            pubsub_stats.get("type").unwrap(),
            &serde_json::Value::String("pubsub".to_string())
        );
        assert_eq!(
            pubsub_stats.get("service_account_configured").unwrap(),
            &serde_json::Value::Bool(true)
        );
        assert!(pubsub_stats.contains_key("config"));
    }

    #[cfg(any(
        not(feature = "kafka"),
        not(feature = "kinesis"),
        not(feature = "google-pubsub")
    ))]
    fn assert_feature_error(result: crate::Result<()>, feature: &str) {
        match result {
            Err(HammerworkError::Streaming { message }) => {
                assert!(
                    message.contains(&format!("`{feature}` feature")),
                    "{message}"
                );
            }
            Err(e) => panic!("unexpected error: {e}"),
            Ok(()) => panic!("stream for a disabled backend must be rejected"),
        }
    }

    #[cfg(not(feature = "kafka"))]
    #[tokio::test]
    async fn test_kafka_stream_rejected_without_feature() {
        let direct = KafkaProcessor::new(vec![], "t".to_string(), HashMap::new()).await;
        assert_feature_error(direct.map(|_| ()), "kafka");

        let manager = StreamManager::new(
            Arc::new(EventManager::new_default()),
            StreamManagerConfig::default(),
        );
        let stream = StreamConfig::new(
            "kafka".to_string(),
            StreamBackend::Kafka {
                brokers: vec!["localhost:9092".to_string()],
                topic: "t".to_string(),
                config: HashMap::new(),
            },
        );
        assert_feature_error(manager.add_stream(stream).await, "kafka");
        assert!(manager.list_streams().await.is_empty());
    }

    #[cfg(not(feature = "kinesis"))]
    #[tokio::test]
    async fn test_kinesis_stream_rejected_without_feature() {
        let manager = StreamManager::new(
            Arc::new(EventManager::new_default()),
            StreamManagerConfig::default(),
        );
        let stream = StreamConfig::new(
            "kinesis".to_string(),
            StreamBackend::Kinesis {
                region: "us-east-1".to_string(),
                stream_name: "s".to_string(),
                access_key_id: None,
                secret_access_key: None,
                config: HashMap::new(),
            },
        );
        assert_feature_error(manager.add_stream(stream).await, "kinesis");
        assert!(manager.list_streams().await.is_empty());
    }

    #[cfg(not(feature = "google-pubsub"))]
    #[tokio::test]
    async fn test_pubsub_stream_rejected_without_feature() {
        let manager = StreamManager::new(
            Arc::new(EventManager::new_default()),
            StreamManagerConfig::default(),
        );
        let stream = StreamConfig::new(
            "pubsub".to_string(),
            StreamBackend::PubSub {
                project_id: "p".to_string(),
                topic_name: "t".to_string(),
                service_account_key: None,
                config: HashMap::new(),
            },
        );
        assert_feature_error(manager.add_stream(stream).await, "google-pubsub");
        assert!(manager.list_streams().await.is_empty());
    }

    fn completed_event() -> JobLifecycleEvent {
        JobLifecycleEvent {
            event_id: Uuid::new_v4(),
            job_id: Uuid::new_v4(),
            queue_name: "shutdown_queue".to_string(),
            event_type: JobLifecycleEventType::Completed,
            priority: JobPriority::Normal,
            timestamp: Utc::now(),
            processing_time_ms: Some(1),
            error: None,
            payload: None,
            metadata: HashMap::new(),
        }
    }

    #[tokio::test]
    async fn test_shutdown_drains_batches_and_stops_processors() {
        let events = Arc::new(EventManager::new_default());
        let manager = StreamManager::new_default(events.clone());
        let stream = StreamConfig {
            buffer_config: BufferConfig {
                batch_size: 1,
                ..Default::default()
            },
            enabled: true,
            ..Default::default()
        };
        let stream_id = stream.id;
        manager
            .add_stream_with_processor(stream, Box::new(InMemoryProcessor::default()))
            .await
            .unwrap();

        events.publish_event(completed_event()).await.unwrap();
        for _ in 0..100 {
            let processed = manager
                .get_stream_stats(stream_id)
                .await
                .is_some_and(|s| s.total_events > 0);
            if processed {
                break;
            }
            tokio::time::sleep(Duration::from_millis(50)).await;
        }

        assert_eq!(manager.shutdown(Duration::from_secs(5)).await, 0);
        assert_eq!(manager.listeners.len(), 0);
        assert!(manager.processors.read().await.is_empty());
        let stats = manager.get_stream_stats(stream_id).await.unwrap();
        assert_eq!(stats.total_events, 1);
        assert_eq!(manager.task_panics(), 0);
    }

    #[tokio::test]
    async fn test_listener_streams_every_event_in_a_burst() {
        let processor = InMemoryProcessor::default();
        let events = Arc::new(EventManager::new_default());
        let manager = StreamManager::new_default(events.clone());
        let stream = StreamConfig {
            buffer_config: BufferConfig {
                batch_size: 10,
                max_buffer_time_secs: 1,
                ..Default::default()
            },
            enabled: true,
            ..Default::default()
        };
        let stream_id = stream.id;
        manager
            .add_stream_with_processor(stream, Box::new(processor.clone()))
            .await
            .unwrap();

        for _ in 0..100 {
            events.publish_event(completed_event()).await.unwrap();
        }

        // Every event published back to back must reach the processor.
        let mut total = 0;
        for _ in 0..100 {
            total = manager
                .get_stream_stats(stream_id)
                .await
                .map_or(0, |s| s.total_events);
            if total >= 100 {
                break;
            }
            tokio::time::sleep(Duration::from_millis(50)).await;
        }
        manager.shutdown(Duration::from_secs(5)).await;
        assert_eq!(total, 100);

        let mut delivered = processor.delivered_ids();
        delivered.sort();
        delivered.dedup();
        assert_eq!(delivered.len(), 100);
        let stats = manager.get_stream_stats(stream_id).await.unwrap();
        assert_eq!(stats.successful_deliveries, 100);
        assert_eq!(stats.failed_deliveries, 0);
    }

    #[tokio::test]
    async fn test_shutdown_aborts_batches_after_grace_period() {
        let manager = StreamManager::new_default(Arc::new(EventManager::new_default()));
        manager.batches.spawn("batch", async {
            tokio::time::sleep(Duration::from_secs(60)).await;
        });
        assert_eq!(manager.shutdown(Duration::from_millis(50)).await, 1);
    }

    #[tokio::test]
    async fn test_panicking_background_task_is_observed() {
        let manager = StreamManager::new_default(Arc::new(EventManager::new_default()));
        manager.batches.spawn("batch", async {
            panic!("expected test panic in a batch task");
        });
        manager.listeners.spawn("listener", async {
            panic!("expected test panic in a listener task");
        });

        // Let both tasks run (and panic) before shutting down.
        for _ in 0..200 {
            if manager.task_panics() == 2 {
                break;
            }
            tokio::time::sleep(Duration::from_millis(10)).await;
        }
        manager.shutdown(Duration::from_secs(5)).await;
        assert_eq!(manager.task_panics(), 2);
    }

    /// A stream that flushes after `batch_size` events and retries per `retry_policy`.
    fn test_stream(batch_size: usize, retry_policy: StreamRetryPolicy) -> StreamConfig {
        StreamConfig {
            buffer_config: BufferConfig {
                batch_size,
                max_buffer_time_secs: 1,
                ..Default::default()
            },
            retry_policy,
            enabled: true,
            ..Default::default()
        }
    }

    /// Retries immediately, without jitter.
    fn immediate_retries(max_attempts: u32) -> StreamRetryPolicy {
        StreamRetryPolicy {
            max_attempts,
            initial_delay_secs: 0,
            max_delay_secs: 0,
            backoff_multiplier: 1.0,
            use_jitter: false,
        }
    }

    /// Add `stream` with `processor`, publish `count` events and wait until the
    /// stream's stats account for all of them. Returns the published event ids.
    async fn stream_events(
        stream: StreamConfig,
        processor: &InMemoryProcessor,
        count: usize,
    ) -> (StreamManager, Uuid, Vec<Uuid>) {
        let events = Arc::new(EventManager::new_default());
        let manager = StreamManager::new_default(events.clone());
        let stream_id = stream.id;
        manager
            .add_stream_with_processor(stream, Box::new(processor.clone()))
            .await
            .unwrap();

        let mut ids = Vec::new();
        for _ in 0..count {
            let event = completed_event();
            ids.push(event.event_id);
            events.publish_event(event).await.unwrap();
        }
        wait_for_total_events(&manager, stream_id, count as u64).await;
        (manager, stream_id, ids)
    }

    async fn wait_for_total_events(manager: &StreamManager, stream_id: Uuid, count: u64) {
        for _ in 0..200 {
            let total = manager
                .get_stream_stats(stream_id)
                .await
                .map_or(0, |s| s.total_events);
            if total >= count {
                return;
            }
            tokio::time::sleep(Duration::from_millis(25)).await;
        }
        panic!("stream did not finish {count} events in time");
    }

    #[tokio::test]
    async fn test_events_reach_the_processor() {
        let processor = InMemoryProcessor::default();
        let (manager, stream_id, ids) =
            stream_events(test_stream(3, immediate_retries(3)), &processor, 3).await;
        manager.shutdown(Duration::from_secs(5)).await;

        let mut delivered = processor.delivered_ids();
        delivered.sort();
        let mut expected = ids;
        expected.sort();
        assert_eq!(delivered, expected);

        let stats = manager.get_stream_stats(stream_id).await.unwrap();
        assert_eq!(stats.total_events, 3);
        assert_eq!(stats.successful_deliveries, 3);
        assert_eq!(stats.failed_deliveries, 0);
        assert_eq!(stats.success_rate, 1.0);
        assert!(stats.last_success_at.is_some());
        assert!(stats.last_failure_at.is_none());
        assert!(stats.last_error.is_none());
    }

    #[tokio::test]
    async fn test_processor_errors_are_retried_then_reported_as_failures() {
        let processor = InMemoryProcessor::default().failing_batches(|_| true);
        let (manager, stream_id, ids) =
            stream_events(test_stream(1, immediate_retries(3)), &processor, 1).await;
        manager.shutdown(Duration::from_secs(5)).await;

        // One initial attempt plus two retries, then the event is given up on.
        assert_eq!(processor.attempts_for(ids[0]), 3);
        assert!(processor.delivered_ids().is_empty());

        let stats = manager.get_stream_stats(stream_id).await.unwrap();
        assert_eq!(stats.total_events, 1);
        assert_eq!(stats.successful_deliveries, 0);
        assert_eq!(stats.failed_deliveries, 1);
        assert_eq!(stats.success_rate, 0.0);
        assert!(stats.last_failure_at.is_some());
        assert!(stats.last_success_at.is_none());
        let last_error = stats.last_error.unwrap();
        assert!(
            last_error.contains("in-memory backend rejected batch 2"),
            "{last_error}"
        );
    }

    #[tokio::test]
    async fn test_transient_error_is_retried_until_delivered() {
        let processor = InMemoryProcessor::default().failing_batches(|call| call == 0);
        let (manager, stream_id, ids) =
            stream_events(test_stream(1, immediate_retries(3)), &processor, 1).await;
        manager.shutdown(Duration::from_secs(5)).await;

        assert_eq!(processor.attempts_for(ids[0]), 2);
        assert_eq!(processor.delivered_ids(), ids);
        let stats = manager.get_stream_stats(stream_id).await.unwrap();
        assert_eq!(stats.successful_deliveries, 1);
        assert_eq!(stats.failed_deliveries, 0);
    }

    #[tokio::test]
    async fn test_partial_batch_failure_retries_only_failed_events() {
        let failing: Arc<std::sync::Mutex<HashSet<Uuid>>> = Arc::default();
        let processor = InMemoryProcessor::default().failing_events({
            let failing = failing.clone();
            move |_, event| failing.lock().unwrap().contains(&event.event.event_id)
        });

        let events = Arc::new(EventManager::new_default());
        let manager = StreamManager::new_default(events.clone());
        let stream = test_stream(4, immediate_retries(2));
        let stream_id = stream.id;
        manager
            .add_stream_with_processor(stream, Box::new(processor.clone()))
            .await
            .unwrap();

        let published: Vec<JobLifecycleEvent> = (0..4).map(|_| completed_event()).collect();
        // The second and fourth events are rejected on every attempt.
        failing
            .lock()
            .unwrap()
            .extend([published[1].event_id, published[3].event_id]);
        for event in &published {
            events.publish_event(event.clone()).await.unwrap();
        }
        wait_for_total_events(&manager, stream_id, 4).await;
        manager.shutdown(Duration::from_secs(5)).await;

        for (i, event) in published.iter().enumerate() {
            let expected_attempts = if i % 2 == 1 { 2 } else { 1 };
            assert_eq!(
                processor.attempts_for(event.event_id),
                expected_attempts,
                "event {i}"
            );
        }
        let mut delivered = processor.delivered_ids();
        delivered.sort();
        let mut expected = vec![published[0].event_id, published[2].event_id];
        expected.sort();
        assert_eq!(delivered, expected);

        let stats = manager.get_stream_stats(stream_id).await.unwrap();
        assert_eq!(stats.total_events, 4);
        assert_eq!(stats.successful_deliveries, 2);
        assert_eq!(stats.failed_deliveries, 2);
        assert_eq!(stats.success_rate, 0.5);
        assert!(
            stats
                .last_error
                .unwrap()
                .contains("in-memory backend rejected event")
        );
    }

    /// A processor that acknowledges only some of the events it was given.
    struct ForgetfulProcessor;

    #[async_trait::async_trait]
    impl StreamProcessor for ForgetfulProcessor {
        async fn send_batch(
            &self,
            _events: Vec<StreamedEvent>,
        ) -> crate::Result<Vec<StreamDelivery>> {
            Ok(Vec::new())
        }

        async fn health_check(&self) -> crate::Result<bool> {
            Ok(true)
        }

        async fn get_stats(&self) -> crate::Result<HashMap<String, serde_json::Value>> {
            Ok(HashMap::new())
        }

        async fn shutdown(&self) -> crate::Result<()> {
            Ok(())
        }
    }

    #[tokio::test]
    async fn test_missing_delivery_result_counts_as_failure() {
        let events = Arc::new(EventManager::new_default());
        let manager = StreamManager::new_default(events.clone());
        let stream = test_stream(1, immediate_retries(1));
        let stream_id = stream.id;
        manager
            .add_stream_with_processor(stream, Box::new(ForgetfulProcessor))
            .await
            .unwrap();
        events.publish_event(completed_event()).await.unwrap();
        wait_for_total_events(&manager, stream_id, 1).await;
        manager.shutdown(Duration::from_secs(5)).await;

        let stats = manager.get_stream_stats(stream_id).await.unwrap();
        assert_eq!(stats.successful_deliveries, 0);
        assert_eq!(stats.failed_deliveries, 1);
        assert!(stats.last_error.unwrap().contains("no delivery result"));
    }

    #[tokio::test]
    async fn test_retries_stop_on_shutdown() {
        let processor = InMemoryProcessor::default().failing_batches(|_| true);
        let events = Arc::new(EventManager::new_default());
        let manager = StreamManager::new_default(events.clone());
        let stream = test_stream(
            1,
            StreamRetryPolicy {
                max_attempts: 5,
                initial_delay_secs: 60,
                max_delay_secs: 60,
                backoff_multiplier: 1.0,
                use_jitter: false,
            },
        );
        let stream_id = stream.id;
        manager
            .add_stream_with_processor(stream, Box::new(processor.clone()))
            .await
            .unwrap();
        events.publish_event(completed_event()).await.unwrap();

        // Wait for the first attempt; the batch is now sleeping before its retry.
        for _ in 0..200 {
            if processor.calls() > 0 {
                break;
            }
            tokio::time::sleep(Duration::from_millis(25)).await;
        }
        assert_eq!(processor.calls(), 1);

        let started = std::time::Instant::now();
        // The batch must finish on its own (not be aborted) well within the grace period.
        assert_eq!(manager.shutdown(Duration::from_secs(30)).await, 0);
        assert!(started.elapsed() < Duration::from_secs(10));
        assert_eq!(processor.calls(), 1);

        let stats = manager.get_stream_stats(stream_id).await.unwrap();
        assert_eq!(stats.total_events, 1);
        assert_eq!(stats.failed_deliveries, 1);
        let last_error = stats.last_error.unwrap();
        assert!(last_error.contains("shut down"), "{last_error}");
        assert!(last_error.contains("rejected batch 0"), "{last_error}");
    }

    #[test]
    fn test_retry_delay_backoff() {
        let policy = StreamRetryPolicy {
            max_attempts: 10,
            initial_delay_secs: 1,
            max_delay_secs: 10,
            backoff_multiplier: 2.0,
            use_jitter: false,
        };
        assert_eq!(policy.delay_for_retry(1), Duration::from_secs(1));
        assert_eq!(policy.delay_for_retry(2), Duration::from_secs(2));
        assert_eq!(policy.delay_for_retry(3), Duration::from_secs(4));
        assert_eq!(policy.delay_for_retry(5), Duration::from_secs(10));
        assert_eq!(policy.delay_for_retry(u32::MAX), Duration::from_secs(10));

        let jittered = StreamRetryPolicy {
            use_jitter: true,
            ..policy.clone()
        };
        for _ in 0..100 {
            let delay = jittered.delay_for_retry(3);
            assert!(delay >= Duration::from_secs(2) && delay <= Duration::from_secs(4));
        }

        let odd = StreamRetryPolicy {
            backoff_multiplier: f64::NAN,
            ..policy
        };
        assert_eq!(odd.delay_for_retry(4), Duration::from_secs(1));
    }

    /// A fake Pub/Sub endpoint that answers every gRPC call with `grpc_status`
    /// (and an empty message when the status is OK), or never answers when
    /// `grpc_status` is `None`.
    #[cfg(feature = "google-pubsub")]
    async fn fake_pubsub_server(grpc_status: Option<u32>) -> String {
        let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
        let addr = listener.local_addr().unwrap().to_string();
        tokio::spawn(async move {
            loop {
                let Ok((socket, _)) = listener.accept().await else {
                    return;
                };
                tokio::spawn(async move {
                    let Ok(mut connection) = h2::server::handshake(socket).await else {
                        return;
                    };
                    while let Some(Ok((_request, mut respond))) = connection.accept().await {
                        let Some(code) = grpc_status else {
                            // Hold the request open without answering.
                            std::mem::forget(respond);
                            continue;
                        };
                        let response = http::Response::builder()
                            .status(200)
                            .header("content-type", "application/grpc")
                            .body(())
                            .unwrap();
                        let Ok(mut stream) = respond.send_response(response, false) else {
                            continue;
                        };
                        if code == 0 {
                            // An empty, uncompressed message: a default `Topic`.
                            let _ = stream.send_data(bytes::Bytes::from_static(&[0; 5]), false);
                        }
                        let mut trailers = http::HeaderMap::new();
                        trailers.insert("grpc-status", code.to_string().parse().unwrap());
                        trailers.insert("grpc-message", "fake".parse().unwrap());
                        let _ = stream.send_trailers(trailers);
                    }
                });
            }
        });
        addr
    }

    #[cfg(feature = "google-pubsub")]
    async fn pubsub_processor_for(addr: String) -> PubSubProcessor {
        let mut config = HashMap::new();
        config.insert("health.check.timeout.ms".to_string(), "500".to_string());
        let client_config = PubSubClientConfig {
            project_id: Some("test-project".to_string()),
            environment: google_cloud_gax::conn::Environment::Emulator(addr),
            ..Default::default()
        };
        PubSubProcessor::with_client_config(
            "test-project".to_string(),
            "test-topic".to_string(),
            None,
            config,
            client_config,
        )
        .await
        .unwrap()
    }

    #[cfg(feature = "google-pubsub")]
    #[tokio::test]
    async fn test_pubsub_health_check_reports_topic_state() {
        // Topic exists.
        let processor = pubsub_processor_for(fake_pubsub_server(Some(0)).await).await;
        assert!(processor.health_check().await.unwrap());

        // NOT_FOUND: the topic does not exist.
        let processor = pubsub_processor_for(fake_pubsub_server(Some(5)).await).await;
        assert!(!processor.health_check().await.unwrap());

        // UNAVAILABLE: Pub/Sub cannot be reached.
        let processor = pubsub_processor_for(fake_pubsub_server(Some(14)).await).await;
        assert!(!processor.health_check().await.unwrap());

        // No answer at all: the health check times out instead of hanging.
        let processor = pubsub_processor_for(fake_pubsub_server(None).await).await;
        let started = std::time::Instant::now();
        assert!(!processor.health_check().await.unwrap());
        assert!(started.elapsed() < Duration::from_secs(5));

        // PERMISSION_DENIED: a configuration problem, reported as an error.
        let processor = pubsub_processor_for(fake_pubsub_server(Some(7)).await).await;
        match processor.health_check().await {
            Err(HammerworkError::Streaming { message }) => {
                assert!(message.contains("rejected"), "{message}");
            }
            other => panic!("expected a streaming error, got {other:?}"),
        }
    }

    /// Disabling a stream stops delivery, enabling it resumes, and removing it drops
    /// its configuration, statistics and subscription. Unknown ids are errors.
    #[tokio::test]
    async fn test_stream_enable_disable_and_remove() {
        let events = Arc::new(EventManager::new_default());
        let manager = StreamManager::new_default(events.clone());
        let processor = InMemoryProcessor::default();
        let stream = test_stream(1, immediate_retries(1));
        let stream_id = stream.id;
        manager
            .add_stream_with_processor(stream, Box::new(processor.clone()))
            .await
            .unwrap();

        assert_eq!(manager.get_stream(stream_id).await.unwrap().id, stream_id);
        assert_eq!(manager.list_streams().await.len(), 1);
        assert_eq!(manager.get_all_stream_stats().await.len(), 1);
        assert_eq!(manager.get_stats().await.active_streams, 1);

        manager.disable_stream(stream_id).await.unwrap();
        assert!(!manager.get_stream(stream_id).await.unwrap().enabled);
        assert_eq!(manager.get_stats().await.active_streams, 0);
        events.publish_event(completed_event()).await.unwrap();
        tokio::time::sleep(Duration::from_millis(300)).await;
        assert!(
            processor.delivered_ids().is_empty(),
            "disabled streams send nothing"
        );

        manager.enable_stream(stream_id).await.unwrap();
        let event = completed_event();
        let event_id = event.event_id;
        events.publish_event(event).await.unwrap();
        wait_for_total_events(&manager, stream_id, 1).await;
        assert_eq!(processor.delivered_ids(), vec![event_id]);

        manager.remove_stream(stream_id).await.unwrap();
        assert!(manager.get_stream(stream_id).await.is_none());
        assert!(manager.get_stream_stats(stream_id).await.is_none());
        assert!(manager.list_streams().await.is_empty());
        events.publish_event(completed_event()).await.unwrap();
        tokio::time::sleep(Duration::from_millis(200)).await;
        assert_eq!(
            processor.delivered_ids().len(),
            1,
            "removed streams send nothing"
        );

        let unknown = Uuid::new_v4();
        for result in [
            manager.enable_stream(unknown).await,
            manager.disable_stream(unknown).await,
        ] {
            let err = result.unwrap_err();
            assert!(err.to_string().contains("not found"), "{err}");
        }
        // Removing an unknown stream is a no-op.
        manager.remove_stream(unknown).await.unwrap();
        manager.shutdown(Duration::from_secs(5)).await;
    }

    #[test]
    fn test_partition_keys_for_job_id_and_hashed_metadata() {
        let mut event = completed_event();
        event
            .metadata
            .insert("tenant".to_string(), "acme".to_string());

        assert_eq!(
            StreamManager::calculate_partition_key(&event, &PartitioningStrategy::JobId),
            Some(event.job_id.to_string())
        );
        assert_eq!(
            StreamManager::calculate_partition_key(
                &event,
                &PartitioningStrategy::Custom {
                    metadata_key: "missing".to_string()
                }
            ),
            None
        );

        let by_tenant = PartitioningStrategy::Hash {
            fields: vec![
                PartitionField::MetadataKey("tenant".to_string()),
                PartitionField::JobId,
            ],
        };
        let key = StreamManager::calculate_partition_key(&event, &by_tenant).unwrap();
        assert!(key.parse::<u32>().unwrap() < 1000);
        assert_eq!(
            StreamManager::calculate_partition_key(&event, &by_tenant),
            Some(key.clone()),
            "deterministic"
        );
        let mut other = event.clone();
        other
            .metadata
            .insert("tenant".to_string(), "globex".to_string());
        assert_ne!(
            StreamManager::calculate_partition_key(&other, &by_tenant),
            Some(key),
            "the metadata value is part of the key"
        );
    }

    /// Wait until `condition` holds, polling every 10 ms for up to 10 seconds.
    async fn eventually<F, Fut>(what: &str, mut condition: F)
    where
        F: FnMut() -> Fut,
        Fut: std::future::Future<Output = bool>,
    {
        for _ in 0..1000 {
            if condition().await {
                return;
            }
            tokio::time::sleep(Duration::from_millis(10)).await;
        }
        panic!("timed out waiting until {what}");
    }

    /// H11/M9: re-adding a stream under the same id (directly, or after removing it)
    /// leaves one listener, and an event published right after `add_stream` returns
    /// is streamed exactly once.
    #[tokio::test]
    async fn test_readding_a_stream_does_not_duplicate_events() {
        let processor = InMemoryProcessor::default();
        let events = Arc::new(EventManager::new_default());
        let manager = StreamManager::new_default(events.clone());
        let stream = test_stream(1, immediate_retries(1));
        let stream_id = stream.id;
        for _ in 0..3 {
            manager
                .add_stream_with_processor(stream.clone(), Box::new(processor.clone()))
                .await
                .unwrap();
        }
        manager.remove_stream(stream_id).await.unwrap();
        manager
            .add_stream_with_processor(stream, Box::new(processor.clone()))
            .await
            .unwrap();
        assert_eq!(events.subscription_count().await, 1);

        let event = completed_event();
        events.publish_event(event.clone()).await.unwrap();
        wait_for_total_events(&manager, stream_id, 1).await;
        tokio::time::sleep(Duration::from_millis(300)).await;
        assert_eq!(processor.delivered_ids(), vec![event.event_id]);
        assert_eq!(processor.calls(), 1);
        manager.shutdown(Duration::from_secs(5)).await;
    }

    /// A processor whose sends wait until `gate` has a permit (one per send).
    #[derive(Clone)]
    struct GatedProcessor {
        gate: Arc<Semaphore>,
        calls: Arc<std::sync::atomic::AtomicUsize>,
    }

    #[async_trait::async_trait]
    impl StreamProcessor for GatedProcessor {
        async fn send_batch(
            &self,
            events: Vec<StreamedEvent>,
        ) -> crate::Result<Vec<StreamDelivery>> {
            self.calls.fetch_add(1, std::sync::atomic::Ordering::SeqCst);
            if let Ok(permit) = self.gate.acquire().await {
                permit.forget();
            }
            Ok(events
                .into_iter()
                .map(|event| StreamDelivery {
                    delivery_id: Uuid::new_v4(),
                    stream_id: Uuid::nil(),
                    event_id: event.event.event_id,
                    success: true,
                    error_message: None,
                    attempted_at: Utc::now(),
                    duration_ms: Some(0),
                    attempt_number: 1,
                    partition: None,
                })
                .collect())
        }

        async fn health_check(&self) -> crate::Result<bool> {
            Ok(true)
        }

        async fn get_stats(&self) -> crate::Result<HashMap<String, serde_json::Value>> {
            Ok(HashMap::new())
        }

        async fn shutdown(&self) -> crate::Result<()> {
            Ok(())
        }
    }

    /// H12: with the backend stuck, at most `max_events` events are in batches; the
    /// listener waits and the overflow is dropped and counted.
    #[tokio::test]
    async fn test_in_flight_events_are_bounded() {
        use std::sync::atomic::Ordering;
        let processor = GatedProcessor {
            gate: Arc::new(Semaphore::new(0)),
            calls: Default::default(),
        };
        let events = Arc::new(EventManager::new(crate::events::EventConfig {
            max_buffer_size: 4,
            ..Default::default()
        }));
        let manager = StreamManager::new_default(events.clone());
        let stream = StreamConfig {
            buffer_config: BufferConfig {
                batch_size: 1,
                max_events: 2,
                max_buffer_time_secs: 1,
            },
            retry_policy: immediate_retries(1),
            ..Default::default()
        };
        let stream_id = stream.id;
        manager
            .add_stream_with_processor(stream, Box::new(processor.clone()))
            .await
            .unwrap();

        for _ in 0..20 {
            events.publish_event(completed_event()).await.unwrap();
            tokio::task::yield_now().await;
        }
        eventually("two batches are being sent", || async {
            processor.calls.load(Ordering::SeqCst) == 2
        })
        .await;
        // The listener is waiting: no third batch is started.
        tokio::time::sleep(Duration::from_millis(200)).await;
        assert_eq!(processor.calls.load(Ordering::SeqCst), 2);
        assert!(manager.batches.len() <= 2, "{}", manager.batches.len());

        // Once the backend recovers, the rest is delivered or counted as dropped.
        processor.gate.add_permits(1_000);
        eventually("every event is delivered or dropped", || async {
            let stats = manager.get_stream_stats(stream_id).await.unwrap();
            stats.successful_deliveries + stats.dropped_events == 20
        })
        .await;
        let stats = manager.get_stream_stats(stream_id).await.unwrap();
        assert!(stats.dropped_events > 0, "{stats:?}");
        manager.shutdown(Duration::from_secs(5)).await;
    }

    /// M4: `global_flush_interval_secs` caps how long a stream buffers an event.
    #[tokio::test]
    async fn test_global_flush_interval_caps_buffer_time() {
        let mut stream = test_stream(100, immediate_retries(1));
        stream.buffer_config.max_buffer_time_secs = 60;
        let config = StreamManagerConfig {
            global_flush_interval_secs: 1,
            ..Default::default()
        };
        assert_eq!(
            StreamManager::flush_after(&stream, &config),
            Duration::from_secs(1)
        );
        assert_eq!(
            StreamManager::flush_after(&stream, &StreamManagerConfig::default()),
            Duration::from_secs(10)
        );

        let processor = InMemoryProcessor::default();
        let events = Arc::new(EventManager::new_default());
        let manager = StreamManager::new(events.clone(), config);
        let stream_id = stream.id;
        manager
            .add_stream_with_processor(stream, Box::new(processor.clone()))
            .await
            .unwrap();
        let started = std::time::Instant::now();
        events.publish_event(completed_event()).await.unwrap();
        wait_for_total_events(&manager, stream_id, 1).await;
        assert!(started.elapsed() < Duration::from_secs(5));
        assert_eq!(processor.delivered_ids().len(), 1);
        manager.shutdown(Duration::from_secs(5)).await;
    }
}