feat: Comprehensive streaming feature flag implementation and serialization testing

- Feature-gate entire streaming module behind streaming/kafka/pubsub/kinesis features
- Wrap all streaming exports, configuration structs, and integration tests with feature flags
- Add comprehensive test coverage for all serialization formats:
  - JSON serialization with round-trip validation
  - MessagePack serialization with size comparison and feature enforcement
  - Avro serialization with binary format validation and feature enforcement
  - Protobuf serialization with efficiency testing and feature enforcement
  - Cross-format size comparison testing
- Ensure graceful degradation when streaming features are disabled
- Maintain backward compatibility while enabling selective feature usage
- All tests pass with and without streaming features enabled

🤖 Generated with [Claude Code](https://claude.ai/code)

Co-Authored-By: Claude <[email protected]>

Author: CodingAnarchy

src/config.rs

@@ -18,6 +18,12 @@ use crate::alerting::AlertingConfig;
 #[cfg(feature = "metrics")]
 use crate::metrics::MetricsConfig;
 
+#[cfg(any(
+    feature = "streaming",
+    feature = "kafka",
+    feature = "google-pubsub",
+    feature = "kinesis"
+))]
 use crate::streaming::StreamConfig;
 use chrono::Duration;
 use serde::{Deserialize, Serialize};
@@ -207,6 +213,12 @@ pub struct HammerworkConfig {
     pub webhooks: WebhookConfigs,
 
     /// Streaming configurations
+    #[cfg(any(
+        feature = "streaming",
+        feature = "kafka",
+        feature = "google-pubsub",
+        feature = "kinesis"
+    ))]
     pub streaming: StreamingConfigs,
 
     /// Alerting configuration
@@ -434,12 +446,18 @@ impl Default for WebhookGlobalSettings {
 }
 
 /// Streaming configurations container
+#[cfg(any(
+    feature = "streaming",
+    feature = "kafka",
+    feature = "google-pubsub",
+    feature = "kinesis"
+))]
 #[derive(Debug, Clone, Serialize, Deserialize, Default)]
 pub struct StreamingConfigs {
     /// List of configured streams
     pub streams: Vec<StreamConfig>,
 
-    /// Global streaming settings
+    /// Global streaming settings  
     pub global_settings: StreamingGlobalSettings,
 }
 
@@ -469,6 +487,12 @@ impl Default for SimpleEventFilter {
 }
 
 /// Global streaming settings
+#[cfg(any(
+    feature = "streaming",
+    feature = "kafka",
+    feature = "google-pubsub",
+    feature = "kinesis"
+))]
 #[derive(Debug, Clone, Serialize, Deserialize)]
 pub struct StreamingGlobalSettings {
     /// Maximum concurrent stream processors
@@ -481,6 +505,12 @@ pub struct StreamingGlobalSettings {
     pub global_flush_interval_secs: u64,
 }
 
+#[cfg(any(
+    feature = "streaming",
+    feature = "kafka",
+    feature = "google-pubsub",
+    feature = "kinesis"
+))]
 impl Default for StreamingGlobalSettings {
     fn default() -> Self {
         Self {
@@ -658,6 +688,12 @@ impl HammerworkConfig {
     }
 
     /// Add a stream configuration
+    #[cfg(any(
+        feature = "streaming",
+        feature = "kafka",
+        feature = "google-pubsub",
+        feature = "kinesis"
+    ))]
     pub fn add_stream(mut self, stream: StreamConfig) -> Self {
         self.streaming.streams.push(stream);
         self
@@ -667,6 +703,12 @@ impl HammerworkConfig {
 #[cfg(test)]
 mod tests {
     use super::*;
+    #[cfg(any(
+        feature = "streaming",
+        feature = "kafka",
+        feature = "google-pubsub",
+        feature = "kinesis"
+    ))]
     use crate::streaming::StreamBackend;
     use tempfile::tempdir;
 
@@ -915,6 +957,12 @@ service_name = "hammerwork"
     }
 
     #[test]
+    #[cfg(any(
+        feature = "streaming",
+        feature = "kafka",
+        feature = "google-pubsub",
+        feature = "kinesis"
+    ))]
     fn test_stream_config() {
         let stream = StreamConfig {
             name: "Test Stream".to_string(),

src/encryption/key_manager.rs

@@ -1228,7 +1228,10 @@ where
 
             // Store the master key securely in the database
             // Note: Database persistence is optional - master key works in-memory only
-            info!("Master key generated and stored in memory with ID: {}", master_key_id);
+            info!(
+                "Master key generated and stored in memory with ID: {}",
+                master_key_id
+            );
 
             // Keep a copy in memory for performance (encrypted with a derived key)
             *self.master_key.lock().map_err(|_| {
@@ -1939,7 +1942,6 @@ where
         }
     }
 
-
     /// Get or create a master key ID based on key material, with database persistence
     async fn get_or_create_master_key_id(
         &self,
@@ -2622,7 +2624,6 @@ impl KeyManager<sqlx::Postgres> {
     pub async fn get_keys_due_for_rotation(&self) -> Result<Vec<String>, EncryptionError> {
         self.get_keys_due_for_rotation_postgres().await
     }
-
 }
 
 #[cfg(feature = "mysql")]
@@ -3144,7 +3145,6 @@ impl KeyManager<sqlx::MySql> {
     pub async fn get_keys_due_for_rotation(&self) -> Result<Vec<String>, EncryptionError> {
         self.get_keys_due_for_rotation_mysql().await
     }
-
 }
 
 impl Default for KeyManagerStats {
@@ -3545,7 +3545,7 @@ mod tests {
                 "rotation-test-key",
                 EncryptionAlgorithm::AES256GCM,
                 KeyPurpose::Encryption,
-                None, // expires_at
+                None,                     // expires_at
                 Some(Duration::days(30)), // 30-day rotation interval
             )
             .await
@@ -3579,7 +3579,7 @@ mod tests {
                 "rotation-test-key",
                 EncryptionAlgorithm::AES256GCM,
                 KeyPurpose::Encryption,
-                None, // expires_at
+                None,                     // expires_at
                 Some(Duration::days(30)), // 30-day rotation interval
             )
             .await
@@ -4166,7 +4166,7 @@ mod tests {
     #[cfg(feature = "encryption")]
     impl TestKeyManager {
         #[allow(dead_code)]
-    fn derive_system_encryption_key(&self, salt: &[u8]) -> Result<Vec<u8>, EncryptionError> {
+        fn derive_system_encryption_key(&self, salt: &[u8]) -> Result<Vec<u8>, EncryptionError> {
             use argon2::{
                 Argon2,
                 password_hash::{PasswordHasher, SaltString},
@@ -4211,7 +4211,7 @@ mod tests {
         }
 
         #[allow(dead_code)]
-    fn encrypt_with_system_key(
+        fn encrypt_with_system_key(
             &self,
             system_key: &[u8],
             plaintext: &[u8],

src/integration_tests.rs

@@ -8,15 +8,22 @@ mod tests {
     use crate::{
         events::{EventFilter, EventManager, JobError, JobLifecycleEvent, JobLifecycleEventType},
         priority::JobPriority,
-        streaming::{
-            PartitioningStrategy, SerializationFormat, StreamBackend, StreamConfig, StreamManager,
-            StreamManagerConfig,
-        },
         webhooks::{
             HttpMethod, RetryPolicy, WebhookAuth, WebhookConfig, WebhookManager,
             WebhookManagerConfig,
         },
     };
+
+    #[cfg(any(
+        feature = "streaming",
+        feature = "kafka",
+        feature = "google-pubsub",
+        feature = "kinesis"
+    ))]
+    use crate::streaming::{
+        PartitioningStrategy, SerializationFormat, StreamBackend, StreamConfig, StreamManager,
+        StreamManagerConfig,
+    };
     use chrono::Utc;
     use std::{collections::HashMap, sync::Arc, time::Duration};
     use tokio::time::sleep;
@@ -110,6 +117,12 @@ mod tests {
     }
 
     #[tokio::test]
+    #[cfg(any(
+        feature = "streaming",
+        feature = "kafka",
+        feature = "google-pubsub",
+        feature = "kinesis"
+    ))]
     async fn test_event_manager_streaming_integration() {
         // Set up event manager
         let event_manager = Arc::new(EventManager::new_default());
@@ -158,6 +171,12 @@ mod tests {
     }
 
     #[tokio::test]
+    #[cfg(any(
+        feature = "streaming",
+        feature = "kafka",
+        feature = "google-pubsub",
+        feature = "kinesis"
+    ))]
     async fn test_full_system_integration() {
         // Set up all components
         let event_manager = Arc::new(EventManager::new_default());
@@ -317,6 +336,12 @@ mod tests {
     }
 
     #[tokio::test]
+    #[cfg(any(
+        feature = "streaming",
+        feature = "kafka",
+        feature = "google-pubsub",
+        feature = "kinesis"
+    ))]
     async fn test_event_filtering_across_systems() {
         let event_manager = Arc::new(EventManager::new_default());
         let webhook_manager =
@@ -435,6 +460,12 @@ mod tests {
     }
 
     #[tokio::test]
+    #[cfg(any(
+        feature = "streaming",
+        feature = "kafka",
+        feature = "google-pubsub",
+        feature = "kinesis"
+    ))]
     async fn test_concurrent_event_processing() {
         let event_manager = Arc::new(EventManager::new_default());
         let webhook_manager = Arc::new(WebhookManager::new(
@@ -528,6 +559,12 @@ mod tests {
     }
 
     #[tokio::test]
+    #[cfg(any(
+        feature = "streaming",
+        feature = "kafka",
+        feature = "google-pubsub",
+        feature = "kinesis"
+    ))]
     async fn test_system_resilience_with_failures() {
         let event_manager = Arc::new(EventManager::new_default());
         let webhook_manager =

src/lib.rs

@@ -258,6 +258,12 @@ pub mod events;
 #[cfg(feature = "webhooks")]
 pub mod webhooks;
 
+#[cfg(any(
+    feature = "streaming",
+    feature = "kafka",
+    feature = "google-pubsub",
+    feature = "kinesis"
+))]
 pub mod streaming;
 
 #[cfg(test)]
@@ -270,8 +276,22 @@ pub use archive::{
 pub use batch::{BatchId, BatchResult, BatchStatus, JobBatch, PartialFailureMode};
 pub use config::{
     ArchiveConfig, DatabaseConfig, HammerworkConfig, LoggingConfig, RateLimitingConfig,
-    StreamingConfigs, StreamingGlobalSettings, WorkerConfig,
+    WorkerConfig,
 };
+
+#[cfg(any(
+    feature = "streaming",
+    feature = "kafka",
+    feature = "google-pubsub",
+    feature = "kinesis"
+))]
+pub use config::{StreamingConfigs, StreamingGlobalSettings};
+#[cfg(any(
+    feature = "streaming",
+    feature = "kafka",
+    feature = "google-pubsub",
+    feature = "kinesis"
+))]
 pub use streaming::StreamConfig;
 
 #[cfg(feature = "webhooks")]
@@ -323,6 +343,12 @@ pub use webhooks::{
     WebhookManager, WebhookManagerConfig, WebhookManagerStats, WebhookStats,
 };
 
+#[cfg(any(
+    feature = "streaming",
+    feature = "kafka",
+    feature = "google-pubsub",
+    feature = "kinesis"
+))]
 pub use streaming::{
     BufferConfig, PartitionField, PartitioningStrategy, SerializationFormat, StreamBackend,
     StreamConfig as StreamSettings, StreamDelivery, StreamManager, StreamManagerConfig,