oxirs_stream/

lib.rs

//! # OxiRS Stream - Ultra-High Performance RDF Streaming Platform
//!
//! [![Version](https://img.shields.io/badge/version-0.1.0-blue)](https://github.com/cool-japan/oxirs/releases)
//! [![docs.rs](https://docs.rs/oxirs-stream/badge.svg)](https://docs.rs/oxirs-stream)
//!
//! **Status**: Production Release (v0.1.0)
//! **Stability**: Public APIs are stable. Production-ready with comprehensive testing.
//!
//! Real-time streaming support with Kafka/NATS/Redis I/O, RDF Patch, SPARQL Update delta,
//! and advanced event processing capabilities.
//!
//! This crate provides enterprise-grade real-time data streaming capabilities for RDF datasets,
//! supporting multiple messaging backends with high-throughput, low-latency guarantees.
//!
//! ## Features
//! - **Multi-Backend Support**: Kafka, NATS JetStream, Redis Streams, AWS Kinesis, Memory
//! - **High Performance**: 100K+ events/second, <10ms latency, exactly-once delivery
//! - **Advanced Event Processing**: Real-time pattern detection, windowing, aggregations
//! - **Enterprise Features**: Circuit breakers, connection pooling, health monitoring
//! - **Standards Compliance**: RDF Patch protocol, SPARQL Update streaming
//!
//! ## Performance Targets
//! - **Throughput**: 100K+ events/second sustained
//! - **Latency**: P99 <10ms for real-time processing
//! - **Reliability**: 99.99% delivery success rate
//! - **Scalability**: Linear scaling to 1000+ partitions

#![allow(dead_code)]

use anyhow::{anyhow, Result};
use chrono::{DateTime, Utc};
use serde::{Deserialize, Serialize};
use std::collections::HashMap;
use std::sync::Arc;
use std::time::{Duration, Instant};
use tokio::sync::{RwLock, Semaphore};
use tracing::{debug, error, info};
use uuid::Uuid;

use crate::backend::StreamBackend;

/// Re-export commonly used types for convenience
pub use backend_optimizer::{
    BackendOptimizer, BackendPerformance, BackendRecommendation, ConsistencyLevel, CostModel,
    OptimizationDecision, OptimizationStats, OptimizerConfig, PatternType, WorkloadPattern,
};
pub use backpressure::{
    BackpressureConfig, BackpressureController, BackpressureStats, BackpressureStrategy,
    FlowControlSignal, RateLimiter as BackpressureRateLimiter,
};
pub use bridge::{
    BridgeInfo, BridgeStatistics, BridgeType, ExternalMessage, ExternalSystemConfig,
    ExternalSystemType, MessageBridgeManager, MessageTransformer, RoutingRule,
};
pub use circuit_breaker::{
    CircuitBreakerError, CircuitBreakerMetrics, FailureType, SharedCircuitBreakerExt,
};
pub use connection_pool::{
    ConnectionFactory, ConnectionPool, DetailedPoolMetrics, LoadBalancingStrategy, PoolConfig,
    PoolStats, PoolStatus,
};
pub use cqrs::{
    CQRSConfig, CQRSHealthStatus, CQRSSystem, Command, CommandBus, CommandBusMetrics,
    CommandHandler, CommandResult, Query, QueryBus, QueryBusMetrics, QueryCacheConfig,
    QueryHandler, QueryResult as CQRSQueryResult, ReadModelManager, ReadModelMetrics,
    ReadModelProjection, RetryConfig as CQRSRetryConfig,
};
pub use delta::{BatchDeltaProcessor, DeltaComputer, DeltaProcessor, ProcessorStats};
pub use dlq::{
    DeadLetterQueue, DlqConfig, DlqEventProcessor, DlqStats as DlqStatsExport, FailedEvent,
    FailureReason,
};
pub use event::{
    EventCategory, EventMetadata, EventPriority, IsolationLevel, QueryResult as EventQueryResult,
    SchemaChangeType, SchemaType, SparqlOperationType, StreamEvent,
};
pub use event_sourcing::{
    EventQuery, EventSnapshot, EventStore, EventStoreConfig, PersistenceBackend, QueryOrder,
    RetentionPolicy, SnapshotConfig, StoredEvent, TimeRange as EventSourcingTimeRange,
};
pub use failover::{ConnectionEndpoint, FailoverConfig, FailoverManager};
pub use graphql_bridge::{
    BridgeConfig, BridgeStats, GraphQLBridge, GraphQLSubscription, GraphQLUpdate,
    GraphQLUpdateType, SubscriptionFilter,
};
pub use multi_region_replication::{
    ConflictResolution, ConflictType, GeographicLocation, MultiRegionReplicationManager,
    RegionConfig, RegionHealth, ReplicatedEvent, ReplicationConfig, ReplicationStats,
    ReplicationStrategy, VectorClock,
};
pub use patch::{PatchParser, PatchSerializer};
pub use performance_optimizer::{
    AdaptiveBatcher, AggregationFunction, AutoTuner, BatchPerformancePoint, BatchSizePredictor,
    BatchingStats, EnhancedMLConfig, MemoryPool, MemoryPoolStats,
    PerformanceConfig as OptimizerPerformanceConfig, ProcessingResult, ProcessingStats,
    ProcessingStatus, TuningDecision, ZeroCopyEvent,
};
pub use schema_registry::{
    CompatibilityMode, ExternalRegistryConfig, RegistryAuth, SchemaDefinition, SchemaFormat,
    SchemaRegistry, SchemaRegistryConfig, ValidationResult, ValidationStats,
};
pub use sparql_streaming::{
    ContinuousQueryManager, QueryManagerConfig, QueryMetadata, QueryResultChannel,
    QueryResultUpdate, UpdateType,
};
pub use store_integration::{
    ChangeDetectionStrategy, ChangeNotification, RealtimeUpdateManager, StoreChangeDetector,
    UpdateChannel, UpdateFilter, UpdateNotification,
};

// Stream, StreamConsumer, and StreamProducer are defined below in this module
pub use biological_computing::{
    AminoAcid, BiologicalProcessingStats, BiologicalStreamProcessor, Cell, CellState,
    CellularAutomaton, ComputationalFunction, DNASequence, EvolutionaryOptimizer, FunctionalDomain,
    Individual, Nucleotide, ProteinStructure, SequenceMetadata,
};
pub use consciousness_streaming::{
    ConsciousnessLevel, ConsciousnessStats, ConsciousnessStreamProcessor, DreamSequence,
    EmotionalContext, IntuitiveEngine, MeditationState,
};
pub use disaster_recovery::{
    BackupCompression, BackupConfig, BackupEncryption, BackupFrequency, BackupJob,
    BackupRetentionPolicy, BackupSchedule, BackupStatus, BackupStorage, BackupType,
    BackupVerification, BackupVerificationResult, BackupWindow, BusinessContinuityConfig,
    ChecksumAlgorithm, CompressionAlgorithm, DRMetrics, DisasterRecoveryConfig,
    DisasterRecoveryManager, DisasterScenario, EncryptionAlgorithm as BackupEncryptionAlgorithm,
    FailoverConfig as DRFailoverConfig, ImpactLevel, KeyDerivationFunction, RecoveryConfig,
    RecoveryOperation, RecoveryPriority, RecoveryRunbook, RecoveryStatus, RecoveryType,
    ReplicationConfig as DRReplicationConfig, ReplicationMode as DRReplicationMode,
    ReplicationTarget as DRReplicationTarget, RunbookExecution, RunbookExecutionStatus,
    RunbookStep, StorageLocation,
};
pub use enterprise_audit::{
    ActionResult, AuditEncryptionConfig, AuditEventType, AuditFilterConfig, AuditMetrics,
    AuditRetentionConfig, AuditSeverity, AuditStorageBackend, AuditStorageConfig,
    AuditStreamingConfig, AuthType, ComplianceConfig, ComplianceFinding, ComplianceReport,
    ComplianceStandard, CompressionType as AuditCompressionType, DestinationAuth, DestinationType,
    EncryptionAlgorithm, EnterpriseAuditConfig, EnterpriseAuditEvent, EnterpriseAuditLogger,
    FindingType, KeyManagementConfig, KmsType, S3AuditConfig, StreamingDestination,
};
pub use enterprise_monitoring::{
    Alert, AlertCondition, AlertManager, AlertRule, AlertSeverity as MonitoringAlertSeverity,
    AlertingConfig, BreachNotificationConfig, ComparisonOperator, EnterpriseMonitoringConfig,
    EnterpriseMonitoringSystem, EscalationLevel, EscalationPolicy, HealthCheckConfig,
    HealthCheckEndpoint, HealthCheckType, MeasurementWindow, MetricDefinition, MetricType,
    MetricValue, MetricsCollector, MetricsConfig, MetricsEndpoint, MetricsEndpointType,
    MetricsExportConfig, MetricsFormat, NotificationChannel, ProfilingConfig, SlaBreach, SlaConfig,
    SlaMeasurement, SlaMetricType, SlaObjective, SlaSeverity, SlaStatus, SlaTracker,
};
pub use multi_tenancy::{
    IsolationMode, MultiTenancyConfig, MultiTenancyManager, MultiTenancyMetrics,
    NamespaceResources, ResourceAllocationStrategy, ResourceType, ResourceUsage, Tenant,
    TenantLifecycleConfig, TenantNamespace, TenantQuota, TenantStatus, TenantTier,
};
pub use observability::{
    AlertConfig, AlertEvent, AlertSeverity, AlertType, BusinessMetrics, SpanLog, SpanStatus,
    StreamObservability, StreamingMetrics, TelemetryConfig, TraceSpan,
};
pub use performance_utils::{
    AdaptiveRateLimiter, IntelligentMemoryPool, IntelligentPrefetcher, ParallelStreamProcessor,
    PerformanceUtilsConfig,
};
pub use quantum_communication::{
    BellState, EntanglementDistribution, QuantumCommConfig, QuantumCommSystem,
    QuantumOperation as QuantumCommOperation, QuantumSecurityProtocol,
    QuantumState as QuantumCommState, Qubit,
};
pub use quantum_streaming::{
    QuantumEvent, QuantumOperation, QuantumProcessingStats, QuantumState, QuantumStreamProcessor,
};
pub use reliability::{BulkReplayResult, DlqStats, ReplayStatus};
pub use rsp::{
    RspConfig, RspLanguage, RspProcessor, RspQuery, StreamClause, StreamDescriptor, Window,
    WindowConfig, WindowSize, WindowStats, WindowType,
};
pub use security::{
    AuditConfig, AuditLogEntry, AuditLogger, AuthConfig, AuthMethod, AuthenticationProvider,
    AuthorizationProvider, AuthzConfig, Credentials, EncryptionConfig, Permission, RateLimitConfig,
    RateLimiter, SecurityConfig as StreamSecurityConfig, SecurityContext, SecurityManager,
    SecurityMetrics, SessionConfig, ThreatAlert, ThreatDetectionConfig, ThreatDetector,
};
pub use temporal_join::{
    IntervalJoin, JoinResult, LateDataConfig, LateDataStrategy, TemporalJoin, TemporalJoinConfig,
    TemporalJoinMetrics, TemporalJoinType, TemporalWindow, TimeSemantics, WatermarkConfig,
    WatermarkStrategy,
};
pub use time_travel::{
    AggregationType, TemporalAggregations, TemporalFilter, TemporalOrdering, TemporalProjection,
    TemporalQuery, TemporalQueryResult, TemporalResultMetadata, TemporalStatistics, TimePoint,
    TimeRange as TimeTravelTimeRange, TimeTravelConfig, TimeTravelEngine, TimeTravelMetrics,
    TimelinePoint,
};
pub use tls_security::{
    CertRotationConfig, CertificateConfig, CertificateFormat, CertificateInfo, CipherSuite,
    ExpiryWarning, MutualTlsConfig, OcspConfig, RevocationCheckConfig, SessionResumptionConfig,
    TlsConfig, TlsManager, TlsMetrics, TlsSessionInfo, TlsVersion,
};
pub use wasm_edge_computing::{
    EdgeExecutionResult, EdgeLocation, OptimizationLevel, PerformanceProfile, PluginCapability,
    PluginSchema, ProcessingSpecialization, ResourceMetrics, SecurityLevel, WasmEdgeConfig,
    WasmEdgeProcessor, WasmPlugin, WasmProcessingResult, WasmProcessorStats, WasmResourceLimits,
};
pub use webhook::{
    EventFilter as WebhookEventFilter, HttpMethod, RateLimit, RetryConfig as WebhookRetryConfig,
    WebhookConfig, WebhookInfo, WebhookManager, WebhookMetadata, WebhookSecurity,
    WebhookStatistics,
};

// New v0.1.0 feature exports
pub use custom_serialization::{
    BenchmarkResults, BsonSerializer, CustomSerializer, FlexBuffersSerializer, IonSerializer,
    RonSerializer, SerializerBenchmark, SerializerBenchmarkSuite, SerializerRegistry,
    SerializerStats, ThriftSerializer,
};
pub use end_to_end_encryption::{
    E2EEConfig, E2EEEncryptionAlgorithm, E2EEManager, E2EEStats, EncryptedMessage,
    HomomorphicEncryption, KeyExchangeAlgorithm, KeyPair, KeyRotationConfig, MultiPartyConfig,
    ZeroKnowledgeProof,
};
pub use gpu_acceleration::{
    AggregationOp, GpuBackend, GpuBuffer, GpuConfig, GpuContext, GpuProcessorConfig, GpuStats,
    GpuStreamProcessor,
};
pub use ml_integration::{
    AnomalyDetectionAlgorithm, AnomalyDetectionConfig, AnomalyDetector, AnomalyResult,
    AnomalyStats, FeatureConfig, FeatureExtractor, FeatureVector, MLIntegrationManager,
    MLModelConfig, ModelMetrics, ModelType, OnlineLearningModel, PredictionResult,
};
pub use rate_limiting::{
    QuotaCheckResult, QuotaEnforcementMode, QuotaLimits, QuotaManager, QuotaOperation,
    RateLimitAlgorithm, RateLimitConfig as AdvancedRateLimitConfig, RateLimitMonitoringConfig,
    RateLimitStats as AdvancedRateLimitStats, RateLimiter as AdvancedRateLimiter,
    RejectionStrategy,
};
pub use scalability::{
    AdaptiveBuffer, AutoScaler, LoadBalancingStrategy as ScalingLoadBalancingStrategy,
    Node as ScalingNode, NodeHealth, Partition, PartitionManager, PartitionStrategy,
    ResourceLimits, ResourceUsage as ScalingResourceUsage, ScalingConfig, ScalingDirection,
    ScalingMode,
};
pub use schema_evolution::{
    CompatibilityCheckResult, CompatibilityIssue, CompatibilityIssueType,
    CompatibilityMode as SchemaCompatibilityMode, DeprecationInfo, EvolutionResult,
    FieldDefinition, FieldType, IssueSeverity, MigrationRule, MigrationStrategy, SchemaChange,
    SchemaDefinition as SchemaEvolutionDefinition, SchemaEvolutionManager,
    SchemaFormat as SchemaEvolutionFormat, SchemaVersion,
};
pub use stream_replay::{
    EventProcessor, ReplayCheckpoint, ReplayConfig, ReplayFilter, ReplayMode, ReplaySpeed,
    ReplayStats, ReplayStatus as StreamReplayStatus, ReplayTransformation, StateSnapshot,
    StreamReplayManager, TransformationType,
};
pub use transactional_processing::{
    IsolationLevel as TransactionalIsolationLevel, LogEntryType, TransactionCheckpoint,
    TransactionLogEntry, TransactionMetadata, TransactionState, TransactionalConfig,
    TransactionalProcessor, TransactionalStats,
};
pub use zero_copy::{
    MemoryMappedBuffer, SharedRefBuffer, SimdBatchProcessor, SimdOperation, SplicedBuffer,
    ZeroCopyBuffer, ZeroCopyConfig, ZeroCopyManager, ZeroCopyStats,
};

// New v0.1.0 exports for developer experience and performance
pub use numa_processing::{
    CpuAffinityMode, HugePageSize, MemoryBandwidthMonitor, MemoryInterleavePolicy, NodeBufferStats,
    NodeProcessorStats, NumaAllocationStrategy, NumaBuffer, NumaBufferPool, NumaBufferPoolConfig,
    NumaBufferPoolStats, NumaConfig, NumaNode, NumaProcessorStats, NumaStreamProcessor,
    NumaThreadPool, NumaThreadPoolStats, NumaTopology, NumaWorker, NumaWorkerStats,
    WorkerDistributionStrategy,
};
pub use out_of_order::{
    EmitStrategy, GapFillingStrategy, LateEventStrategy, OrderedEvent, OutOfOrderConfig,
    OutOfOrderHandler, OutOfOrderHandlerBuilder, OutOfOrderStats, SequenceTracker, Watermark,
};
pub use performance_profiler::{
    HistogramStats, LatencyHistogram, OperationTimer, PerformanceProfiler, PerformanceReport,
    PerformanceSample, PerformanceWarning, ProfilerBuilder, ProfilerConfig, ProfilerStats,
    Recommendation, RecommendationCategory, RecommendationEffort, RecommendationImpact, Span,
    WarningSeverity, WarningThresholds, WarningType,
};
pub use stream_sql::{
    AggregateFunction, BinaryOperator, ColumnDefinition, CreateStreamStatement, DataType,
    Expression, FromClause, JoinType, Lexer, OrderByItem, Parser,
    QueryResult as StreamSqlQueryResult, QueryType, ResultRow, SelectItem, SelectStatement,
    SqlValue, StreamMetadata, StreamSqlConfig, StreamSqlEngine, StreamSqlStats, Token,
    UnaryOperator, WindowSpec, WindowType as SqlWindowType,
};
pub use testing_framework::{
    Assertion, AssertionType, CapturedEvent, EventGenerator, EventMatcher, GeneratorConfig,
    GeneratorType, MockClock, PerformanceMetric, TestFixture, TestHarness, TestHarnessBuilder,
    TestHarnessConfig, TestMetrics, TestReport, TestStatus,
};

// New v0.1.0 exports for ML, versioning, and migration
pub use anomaly_detection::{
    Anomaly, AnomalyAlert, AnomalyConfig, AnomalyDetector as AdaptiveAnomalyDetector,
    AnomalySeverity, AnomalyStats as AdaptiveAnomalyStats, DetectorType, MultiDimensionalDetector,
};
pub use migration_tools::{
    APIMapping, ConceptMapping, GeneratedFile, GeneratedFileType, ManualReviewItem,
    MigrationConfig, MigrationError, MigrationReport, MigrationSuggestion, MigrationTool,
    MigrationWarning, QuickStart, ReviewPriority, SourcePlatform, SuggestionCategory,
};
pub use online_learning::{
    ABTestConfig, ABTestResult, DriftDetection, ModelCheckpoint,
    ModelMetrics as OnlineModelMetrics, ModelType as OnlineModelType, OnlineLearningConfig,
    OnlineLearningModel as StreamOnlineLearningModel, OnlineLearningStats, Prediction, Sample,
    StreamFeatureExtractor,
};
pub use stream_versioning::{
    Branch, BranchId, Change, ChangeType, Changeset, Snapshot, StreamVersioning, TimeTravelQuery,
    TimeTravelTarget, VersionDiff, VersionId, VersionMetadata, VersionedEvent, VersioningConfig,
    VersioningStats,
};

// New v0.1.0 advanced ML exports
pub use automl_stream::{
    Algorithm, AutoML, AutoMLConfig, AutoMLStats, HyperParameters, ModelPerformance, TaskType,
    TrainedModel,
};
pub use feature_engineering::{
    Feature, FeatureExtractionConfig, FeatureMetadata, FeaturePipeline, FeatureSet, FeatureStore,
    FeatureTransform, FeatureValue, ImputationStrategy, PipelineStats,
};
pub use neural_architecture_search::{
    ActivationType, Architecture, ArchitecturePerformance, LayerType, NASConfig, NASStats,
    ObjectiveWeights, SearchSpace, SearchStrategy, NAS,
};
pub use predictive_analytics::{
    AccuracyMetrics, ForecastAlgorithm, ForecastResult, ForecastingConfig, PredictiveAnalytics,
    PredictiveStats, SeasonalityType, TrendDirection,
};
pub use reinforcement_learning::{
    Action, Experience, RLAgent, RLAlgorithm, RLConfig, RLStats, State as RLState,
};

// Utility exports
pub use utils::{
    create_dev_stream, create_prod_stream, BatchProcessor, EventFilter, EventSampler,
    SimpleRateLimiter, StreamMultiplexer, StreamStats,
};

// Advanced SciRS2 optimization exports
pub use advanced_scirs2_optimization::{
    AdvancedOptimizerConfig, AdvancedStreamOptimizer, MovingStats, OptimizerMetrics,
};
pub use cdc_processor::{
    CdcConfig, CdcConnector, CdcEvent, CdcEventBuilder, CdcMetrics, CdcOperation, CdcProcessor,
    CdcSource,
};

// Adaptive load shedding exports
pub use adaptive_load_shedding::{
    DropStrategy, LoadMetrics, LoadSheddingConfig, LoadSheddingManager, LoadSheddingStats,
};

// Stream fusion optimizer exports
pub use stream_fusion::{
    FusableChain, FusedOperation, FusedType, FusionAnalysis, FusionConfig, FusionOptimizer,
    FusionStats, Operation,
};

// Complex Event Processing (CEP) engine exports
pub use cep_engine::{
    CepAggregationFunction, CepConfig, CepEngine, CepMetrics, CepStatistics, CompleteMatch,
    CorrelationFunction, CorrelationResult, CorrelationStats, DetectedPattern, DetectionAlgorithm,
    DetectionStats, EnrichmentData, EnrichmentService, EnrichmentSource, EnrichmentSourceType,
    EnrichmentStats, EventBuffer, EventCorrelator, EventPattern, FieldPredicate, PartialMatch,
    PatternDetector, ProcessingRule, RuleAction, RuleCondition, RuleEngine, RuleExecutionStats,
    State, StateMachine, TemporalOperator, TimestampedEvent,
};

// Data quality and validation framework exports
pub use data_quality::{
    AlertCondition as QualityAlertCondition, AlertManager as QualityAlertManager,
    AlertRule as QualityAlertRule, AlertSeverity as QualityAlertSeverity,
    AlertStats as QualityAlertStats, AlertType as QualityAlertType, AuditAction, AuditEntry,
    AuditStats, AuditTrail, CleansingRule, CleansingStats, CorrectionType, DataCleanser,
    DataCorrection, DataProfiler, DataQualityValidator, DuplicateDetector, DuplicateStats,
    FailureSeverity, FieldProfile, OutlierMethod, ProfileStats, ProfiledEvent, QualityAlert,
    QualityConfig, QualityDimension, QualityMetrics, QualityReport, QualityScorer, ScoringStats,
    ValidationFailure, ValidationResult as QualityValidationResult, ValidationRule,
};

// Advanced sampling techniques exports
pub use advanced_sampling::{
    AdvancedSamplingManager, BloomFilter, BloomFilterStats, CountMinSketch, CountMinSketchStats,
    HyperLogLog, HyperLogLogStats, ReservoirSampler, ReservoirStats, SamplingConfig,
    SamplingManagerStats, StratifiedSampler, StratifiedStats, TDigest, TDigestStats,
};

pub mod backend;
pub mod backend_optimizer;
pub mod backpressure;
pub mod biological_computing;
pub mod bridge;
pub mod circuit_breaker;
pub mod config;
pub mod connection_pool;
pub mod consciousness_streaming;
pub mod consumer;
pub mod cqels;
pub mod cqrs;
pub mod csparql;
pub mod delta;
pub mod diagnostics;
pub mod disaster_recovery;
pub mod dlq;
pub mod enterprise_audit;
pub mod enterprise_monitoring;
pub mod error;
pub mod event;
pub mod event_sourcing;
pub mod failover;
pub mod graphql_bridge;
pub mod graphql_subscriptions;
pub mod health_monitor;
pub mod join;
pub mod monitoring;
pub mod multi_region_replication;
pub mod multi_tenancy;
pub mod observability;
pub mod patch;
pub mod performance_optimizer;
pub mod performance_utils;
pub mod processing;
pub mod producer;
pub mod quantum_communication;
pub mod quantum_processing;
pub mod quantum_streaming;
pub mod reconnect;
pub mod reliability;
pub mod rsp;
pub mod schema_registry;
pub mod security;
pub mod serialization;
pub mod sparql_streaming;
pub mod state;
pub mod store_integration;
pub mod temporal_join;
pub mod time_travel;
pub mod tls_security;
pub mod types;
pub mod wasm_edge_computing;
pub mod webhook;

// New v0.1.0 modules for advanced features
pub mod custom_serialization;
pub mod end_to_end_encryption;
pub mod gpu_acceleration;
pub mod ml_integration;
pub mod rate_limiting;
pub mod scalability;
pub mod schema_evolution;
pub mod stream_replay;
pub mod transactional_processing;
pub mod zero_copy;

// New v0.1.0 modules for developer experience and performance
pub mod numa_processing;
pub mod out_of_order;
pub mod performance_profiler;
pub mod stream_sql;
pub mod testing_framework;

// New v0.1.0 modules for ML, versioning, and migration
pub mod anomaly_detection;
pub mod migration_tools;
pub mod online_learning;
pub mod stream_versioning;

// Advanced ML modules for v0.1.0 completion
pub mod automl_stream;
pub mod feature_engineering;
pub mod neural_architecture_search;
pub mod predictive_analytics;
pub mod reinforcement_learning;

// Utilities module
pub mod utils;

// Advanced SciRS2 optimization module
pub mod advanced_scirs2_optimization;
pub mod cdc_processor;

// Adaptive load shedding module
pub mod adaptive_load_shedding;

// Stream fusion optimizer module
pub mod stream_fusion;

// Complex Event Processing (CEP) engine module
pub mod cep_engine;

// Data quality and validation framework module
pub mod data_quality;

// Advanced sampling techniques module
pub mod advanced_sampling;

/// Enhanced stream configuration with advanced features
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct StreamConfig {
    pub backend: StreamBackendType,
    pub topic: String,
    pub batch_size: usize,
    pub flush_interval_ms: u64,
    /// Maximum concurrent connections
    pub max_connections: usize,
    /// Connection timeout
    pub connection_timeout: Duration,
    /// Enable compression
    pub enable_compression: bool,
    /// Compression type
    pub compression_type: CompressionType,
    /// Retry configuration
    pub retry_config: RetryConfig,
    /// Circuit breaker configuration
    pub circuit_breaker: CircuitBreakerConfig,
    /// Security configuration
    pub security: SecurityConfig,
    /// Performance tuning
    pub performance: StreamPerformanceConfig,
    /// Monitoring configuration
    pub monitoring: MonitoringConfig,
}

/// Compression types supported
#[derive(Debug, Clone, Serialize, Deserialize)]
pub enum CompressionType {
    None,
    Gzip,
    Snappy,
    Lz4,
    Zstd,
}

/// Retry configuration
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct RetryConfig {
    pub max_retries: u32,
    pub initial_backoff: Duration,
    pub max_backoff: Duration,
    pub backoff_multiplier: f64,
    pub jitter: bool,
}

/// Circuit breaker configuration
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct CircuitBreakerConfig {
    pub enabled: bool,
    pub failure_threshold: u32,
    pub success_threshold: u32,
    pub timeout: Duration,
    pub half_open_max_calls: u32,
}

/// Security configuration
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct SecurityConfig {
    pub enable_tls: bool,
    pub verify_certificates: bool,
    pub client_cert_path: Option<String>,
    pub client_key_path: Option<String>,
    pub ca_cert_path: Option<String>,
    pub sasl_config: Option<SaslConfig>,
}

/// SASL authentication configuration
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct SaslConfig {
    pub mechanism: SaslMechanism,
    pub username: String,
    pub password: String,
}

/// SASL authentication mechanisms
#[derive(Debug, Clone, Serialize, Deserialize)]
pub enum SaslMechanism {
    Plain,
    ScramSha256,
    ScramSha512,
    OAuthBearer,
}

/// Performance tuning configuration
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct StreamPerformanceConfig {
    pub enable_batching: bool,
    pub enable_pipelining: bool,
    pub buffer_size: usize,
    pub prefetch_count: u32,
    pub enable_zero_copy: bool,
    pub enable_simd: bool,
    pub parallel_processing: bool,
    pub worker_threads: Option<usize>,
}

/// Monitoring configuration
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct MonitoringConfig {
    pub enable_metrics: bool,
    pub enable_tracing: bool,
    pub metrics_interval: Duration,
    pub health_check_interval: Duration,
    pub enable_profiling: bool,
    pub prometheus_endpoint: Option<String>,
    pub jaeger_endpoint: Option<String>,
    pub log_level: String,
}

/// Enhanced streaming backend options
#[derive(Debug, Clone, Serialize, Deserialize)]
pub enum StreamBackendType {
    #[cfg(feature = "kafka")]
    Kafka {
        brokers: Vec<String>,
        security_protocol: Option<String>,
        sasl_config: Option<SaslConfig>,
    },
    #[cfg(feature = "nats")]
    Nats {
        url: String,
        cluster_urls: Option<Vec<String>>,
        jetstream_config: Option<NatsJetStreamConfig>,
    },
    #[cfg(feature = "redis")]
    Redis {
        url: String,
        cluster_urls: Option<Vec<String>>,
        pool_size: Option<usize>,
    },
    #[cfg(feature = "kinesis")]
    Kinesis {
        region: String,
        stream_name: String,
        credentials: Option<AwsCredentials>,
    },
    #[cfg(feature = "pulsar")]
    Pulsar {
        service_url: String,
        auth_config: Option<PulsarAuthConfig>,
    },
    #[cfg(feature = "rabbitmq")]
    RabbitMQ {
        url: String,
        exchange: Option<String>,
        queue: Option<String>,
    },
    Memory {
        max_size: Option<usize>,
        persistence: bool,
    },
}

/// NATS JetStream configuration
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct NatsJetStreamConfig {
    pub domain: Option<String>,
    pub api_prefix: Option<String>,
    pub timeout: Duration,
}

/// AWS credentials configuration
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct AwsCredentials {
    pub access_key_id: String,
    pub secret_access_key: String,
    pub session_token: Option<String>,
    pub role_arn: Option<String>,
}

/// Pulsar authentication configuration
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct PulsarAuthConfig {
    pub auth_method: PulsarAuthMethod,
    pub auth_params: HashMap<String, String>,
}

/// Pulsar authentication methods
#[derive(Debug, Clone, Serialize, Deserialize)]
pub enum PulsarAuthMethod {
    Token,
    Jwt,
    Oauth2,
    Tls,
}

/// Enhanced stream producer for publishing RDF changes with backend support
pub struct StreamProducer {
    config: StreamConfig,
    backend_producer: BackendProducer,
    stats: Arc<RwLock<ProducerStats>>,
    circuit_breaker: Option<circuit_breaker::SharedCircuitBreaker>,
    last_flush: Instant,
    _pending_events: Arc<RwLock<Vec<StreamEvent>>>,
    batch_buffer: Arc<RwLock<Vec<StreamEvent>>>,
    flush_semaphore: Arc<Semaphore>,
}

/// Backend-agnostic producer wrapper
enum BackendProducer {
    #[cfg(feature = "kafka")]
    Kafka(backend::kafka::KafkaProducer),
    #[cfg(feature = "nats")]
    Nats(Box<backend::nats::NatsProducer>),
    #[cfg(feature = "redis")]
    Redis(backend::redis::RedisProducer),
    #[cfg(feature = "kinesis")]
    Kinesis(backend::kinesis::KinesisProducer),
    #[cfg(feature = "pulsar")]
    Pulsar(Box<backend::pulsar::PulsarProducer>),
    #[cfg(feature = "rabbitmq")]
    RabbitMQ(Box<backend::rabbitmq::RabbitMQProducer>),
    Memory(MemoryProducer),
}

/// Producer statistics for monitoring
#[derive(Debug, Default, Clone)]
pub struct ProducerStats {
    events_published: u64,
    events_failed: u64,
    _bytes_sent: u64,
    avg_latency_ms: f64,
    max_latency_ms: u64,
    batch_count: u64,
    flush_count: u64,
    circuit_breaker_trips: u64,
    last_publish: Option<DateTime<Utc>>,
    backend_type: String,
}

/// Memory-based producer for testing and development
// Type aliases for complex types
type MemoryEventVec = Vec<(DateTime<Utc>, StreamEvent)>;
type MemoryEventStore = Arc<RwLock<MemoryEventVec>>;

// Global shared storage for memory backend events
static MEMORY_EVENTS: std::sync::OnceLock<MemoryEventStore> = std::sync::OnceLock::new();

pub fn get_memory_events() -> MemoryEventStore {
    MEMORY_EVENTS
        .get_or_init(|| Arc::new(RwLock::new(Vec::new())))
        .clone()
}

/// Clear the global memory storage (for testing)
pub async fn clear_memory_events() {
    let events = get_memory_events();
    events.write().await.clear();
    // Also clear the memory backend storage
    backend::memory::clear_memory_storage().await;
}

struct MemoryProducer {
    backend: Box<dyn StreamBackend + Send + Sync>,
    topic: String,
    stats: ProducerStats,
}

impl MemoryProducer {
    fn _new(_max_size: Option<usize>, _persistence: bool) -> Self {
        Self {
            backend: Box::new(backend::memory::MemoryBackend::new()),
            topic: "oxirs-stream".to_string(), // Default topic
            stats: ProducerStats {
                backend_type: "memory".to_string(),
                ..Default::default()
            },
        }
    }

    fn with_topic(topic: String) -> Self {
        Self {
            backend: Box::new(backend::memory::MemoryBackend::new()),
            topic,
            stats: ProducerStats {
                backend_type: "memory".to_string(),
                ..Default::default()
            },
        }
    }

    async fn publish(&mut self, event: StreamEvent) -> Result<()> {
        let start_time = Instant::now();

        // Use the backend implementation
        self.backend
            .as_mut()
            .connect()
            .await
            .map_err(|e| anyhow!("Backend connect failed: {}", e))?;

        // Create topic if it doesn't exist
        let topic_name = crate::types::TopicName::new(self.topic.clone());
        self.backend
            .create_topic(&topic_name, 1)
            .await
            .map_err(|e| anyhow!("Topic creation failed: {}", e))?;

        // Send event through backend
        self.backend
            .send_event(&topic_name, event)
            .await
            .map_err(|e| anyhow!("Send event failed: {}", e))?;

        // Update stats
        self.stats.events_published += 1;
        let latency = start_time.elapsed().as_millis() as u64;
        self.stats.max_latency_ms = self.stats.max_latency_ms.max(latency);
        self.stats.avg_latency_ms = (self.stats.avg_latency_ms + latency as f64) / 2.0;
        self.stats.last_publish = Some(Utc::now());

        debug!("Memory producer: published event via backend");
        Ok(())
    }

    async fn flush(&mut self) -> Result<()> {
        // Backend handles flushing internally
        self.stats.flush_count += 1;
        debug!("Memory producer: flush completed");
        Ok(())
    }

    fn _get_stats(&self) -> &ProducerStats {
        &self.stats
    }
}

impl StreamProducer {
    /// Create a new enhanced stream producer with backend support
    pub async fn new(config: StreamConfig) -> Result<Self> {
        // Initialize circuit breaker if enabled
        let circuit_breaker = if config.circuit_breaker.enabled {
            Some(circuit_breaker::new_shared_circuit_breaker(
                circuit_breaker::CircuitBreakerConfig {
                    enabled: config.circuit_breaker.enabled,
                    failure_threshold: config.circuit_breaker.failure_threshold,
                    success_threshold: config.circuit_breaker.success_threshold,
                    timeout: config.circuit_breaker.timeout,
                    half_open_max_calls: config.circuit_breaker.half_open_max_calls,
                    ..Default::default()
                },
            ))
        } else {
            None
        };

        // Initialize backend-specific producer
        let backend_producer = match &config.backend {
            #[cfg(feature = "kafka")]
            StreamBackendType::Kafka {
                brokers,
                security_protocol,
                sasl_config,
            } => {
                let stream_config = crate::StreamConfig {
                    backend: crate::StreamBackendType::Kafka {
                        brokers: brokers.clone(),
                        security_protocol: security_protocol.clone(),
                        sasl_config: sasl_config.clone(),
                    },
                    topic: config.topic.clone(),
                    batch_size: config.batch_size,
                    flush_interval_ms: config.flush_interval_ms,
                    max_connections: config.max_connections,
                    connection_timeout: config.connection_timeout,
                    enable_compression: config.enable_compression,
                    compression_type: config.compression_type.clone(),
                    retry_config: config.retry_config.clone(),
                    circuit_breaker: config.circuit_breaker.clone(),
                    security: config.security.clone(),
                    performance: config.performance.clone(),
                    monitoring: config.monitoring.clone(),
                };

                let mut producer = backend::kafka::KafkaProducer::new(stream_config)?;
                producer.connect().await?;
                BackendProducer::Kafka(producer)
            }
            #[cfg(feature = "nats")]
            StreamBackendType::Nats {
                url,
                cluster_urls,
                jetstream_config,
            } => {
                let stream_config = crate::StreamConfig {
                    backend: crate::StreamBackendType::Nats {
                        url: url.clone(),
                        cluster_urls: cluster_urls.clone(),
                        jetstream_config: jetstream_config.clone(),
                    },
                    topic: config.topic.clone(),
                    batch_size: config.batch_size,
                    flush_interval_ms: config.flush_interval_ms,
                    max_connections: config.max_connections,
                    connection_timeout: config.connection_timeout,
                    enable_compression: config.enable_compression,
                    compression_type: config.compression_type.clone(),
                    retry_config: config.retry_config.clone(),
                    circuit_breaker: config.circuit_breaker.clone(),
                    security: config.security.clone(),
                    performance: config.performance.clone(),
                    monitoring: config.monitoring.clone(),
                };

                let mut producer = backend::nats::NatsProducer::new(stream_config)?;
                producer.connect().await?;
                BackendProducer::Nats(Box::new(producer))
            }
            #[cfg(feature = "redis")]
            StreamBackendType::Redis {
                url,
                cluster_urls,
                pool_size,
            } => {
                let stream_config = crate::StreamConfig {
                    backend: crate::StreamBackendType::Redis {
                        url: url.clone(),
                        cluster_urls: cluster_urls.clone(),
                        pool_size: *pool_size,
                    },
                    topic: config.topic.clone(),
                    batch_size: config.batch_size,
                    flush_interval_ms: config.flush_interval_ms,
                    max_connections: config.max_connections,
                    connection_timeout: config.connection_timeout,
                    enable_compression: config.enable_compression,
                    compression_type: config.compression_type.clone(),
                    retry_config: config.retry_config.clone(),
                    circuit_breaker: config.circuit_breaker.clone(),
                    security: config.security.clone(),
                    performance: config.performance.clone(),
                    monitoring: config.monitoring.clone(),
                };

                let mut producer = backend::redis::RedisProducer::new(stream_config)?;
                producer.connect().await?;
                BackendProducer::Redis(producer)
            }
            #[cfg(feature = "kinesis")]
            StreamBackendType::Kinesis {
                region,
                stream_name,
                credentials,
            } => {
                let stream_config = crate::StreamConfig {
                    backend: crate::StreamBackendType::Kinesis {
                        region: region.clone(),
                        stream_name: stream_name.clone(),
                        credentials: credentials.clone(),
                    },
                    topic: config.topic.clone(),
                    batch_size: config.batch_size,
                    flush_interval_ms: config.flush_interval_ms,
                    max_connections: config.max_connections,
                    connection_timeout: config.connection_timeout,
                    enable_compression: config.enable_compression,
                    compression_type: config.compression_type.clone(),
                    retry_config: config.retry_config.clone(),
                    circuit_breaker: config.circuit_breaker.clone(),
                    security: config.security.clone(),
                    performance: config.performance.clone(),
                    monitoring: config.monitoring.clone(),
                };

                let mut producer = backend::kinesis::KinesisProducer::new(stream_config)?;
                producer.connect().await?;
                BackendProducer::Kinesis(producer)
            }
            #[cfg(feature = "pulsar")]
            StreamBackendType::Pulsar {
                service_url,
                auth_config,
            } => {
                let stream_config = crate::StreamConfig {
                    backend: crate::StreamBackendType::Pulsar {
                        service_url: service_url.clone(),
                        auth_config: auth_config.clone(),
                    },
                    topic: config.topic.clone(),
                    batch_size: config.batch_size,
                    flush_interval_ms: config.flush_interval_ms,
                    max_connections: config.max_connections,
                    connection_timeout: config.connection_timeout,
                    enable_compression: config.enable_compression,
                    compression_type: config.compression_type.clone(),
                    retry_config: config.retry_config.clone(),
                    circuit_breaker: config.circuit_breaker.clone(),
                    security: config.security.clone(),
                    performance: config.performance.clone(),
                    monitoring: config.monitoring.clone(),
                };

                let mut producer = backend::pulsar::PulsarProducer::new(stream_config)?;
                producer.connect().await?;
                BackendProducer::Pulsar(Box::new(producer))
            }
            #[cfg(feature = "rabbitmq")]
            StreamBackendType::RabbitMQ {
                url,
                exchange,
                queue,
            } => {
                let stream_config = crate::StreamConfig {
                    backend: crate::StreamBackendType::RabbitMQ {
                        url: url.clone(),
                        exchange: exchange.clone(),
                        queue: queue.clone(),
                    },
                    topic: config.topic.clone(),
                    batch_size: config.batch_size,
                    flush_interval_ms: config.flush_interval_ms,
                    max_connections: config.max_connections,
                    connection_timeout: config.connection_timeout,
                    enable_compression: config.enable_compression,
                    compression_type: config.compression_type.clone(),
                    retry_config: config.retry_config.clone(),
                    circuit_breaker: config.circuit_breaker.clone(),
                    security: config.security.clone(),
                    performance: config.performance.clone(),
                    monitoring: config.monitoring.clone(),
                };

                let mut producer = backend::rabbitmq::RabbitMQProducer::new(stream_config)?;
                producer.connect().await?;
                BackendProducer::RabbitMQ(Box::new(producer))
            }
            StreamBackendType::Memory {
                max_size: _,
                persistence: _,
            } => BackendProducer::Memory(MemoryProducer::with_topic(config.topic.clone())),
        };

        let stats = Arc::new(RwLock::new(ProducerStats {
            backend_type: match backend_producer {
                #[cfg(feature = "kafka")]
                BackendProducer::Kafka(_) => "kafka".to_string(),
                #[cfg(feature = "nats")]
                BackendProducer::Nats(_) => "nats".to_string(),
                #[cfg(feature = "redis")]
                BackendProducer::Redis(_) => "redis".to_string(),
                #[cfg(feature = "kinesis")]
                BackendProducer::Kinesis(_) => "kinesis".to_string(),
                #[cfg(feature = "pulsar")]
                BackendProducer::Pulsar(_) => "pulsar".to_string(),
                #[cfg(feature = "rabbitmq")]
                BackendProducer::RabbitMQ(_) => "rabbitmq".to_string(),
                BackendProducer::Memory(_) => "memory".to_string(),
            },
            ..Default::default()
        }));

        info!(
            "Created stream producer with backend: {}",
            stats.read().await.backend_type
        );

        Ok(Self {
            config,
            backend_producer,
            stats,
            circuit_breaker,
            last_flush: Instant::now(),
            _pending_events: Arc::new(RwLock::new(Vec::new())),
            batch_buffer: Arc::new(RwLock::new(Vec::new())),
            flush_semaphore: Arc::new(Semaphore::new(1)),
        })
    }

    /// Publish a stream event with circuit breaker protection and batching
    pub async fn publish(&mut self, event: StreamEvent) -> Result<()> {
        let start_time = Instant::now();

        // Check circuit breaker if enabled
        if let Some(cb) = &self.circuit_breaker {
            if !cb.can_execute().await {
                self.stats.write().await.circuit_breaker_trips += 1;
                return Err(anyhow!("Circuit breaker is open - cannot publish events"));
            }
        }

        // Handle batching if enabled
        if self.config.performance.enable_batching {
            let mut batch_buffer = self.batch_buffer.write().await;
            batch_buffer.push(event);

            if batch_buffer.len() >= self.config.batch_size {
                let events = std::mem::take(&mut *batch_buffer);
                drop(batch_buffer);
                return self.publish_batch_internal(events).await;
            }

            return Ok(());
        }

        // Publish single event
        let result = self.publish_single_event(event).await;

        // Update circuit breaker and stats
        match &result {
            Ok(_) => {
                if let Some(cb) = &self.circuit_breaker {
                    cb.record_success_with_duration(start_time.elapsed()).await;
                }

                let mut stats = self.stats.write().await;
                stats.events_published += 1;
                let latency = start_time.elapsed().as_millis() as u64;
                stats.max_latency_ms = stats.max_latency_ms.max(latency);
                stats.avg_latency_ms = (stats.avg_latency_ms + latency as f64) / 2.0;
                stats.last_publish = Some(Utc::now());
            }
            Err(_) => {
                if let Some(cb) = &self.circuit_breaker {
                    cb.record_failure_with_type(circuit_breaker::FailureType::NetworkError)
                        .await;
                }

                self.stats.write().await.events_failed += 1;
            }
        }

        result
    }

    /// Publish a single event to the backend
    async fn publish_single_event(&mut self, event: StreamEvent) -> Result<()> {
        match &mut self.backend_producer {
            #[cfg(feature = "kafka")]
            BackendProducer::Kafka(producer) => producer.publish(event).await,
            #[cfg(feature = "nats")]
            BackendProducer::Nats(producer) => producer.publish(event).await,
            #[cfg(feature = "redis")]
            BackendProducer::Redis(producer) => producer.publish(event).await,
            #[cfg(feature = "kinesis")]
            BackendProducer::Kinesis(producer) => producer.publish(event).await,
            #[cfg(feature = "pulsar")]
            BackendProducer::Pulsar(producer) => producer.publish(event).await,
            #[cfg(feature = "rabbitmq")]
            BackendProducer::RabbitMQ(producer) => producer.publish(event).await,
            BackendProducer::Memory(producer) => producer.publish(event).await,
        }
    }

    /// Publish multiple events as a batch
    pub async fn publish_batch(&mut self, events: Vec<StreamEvent>) -> Result<()> {
        if events.is_empty() {
            return Ok(());
        }

        self.publish_batch_internal(events).await
    }

    /// Internal batch publishing implementation
    async fn publish_batch_internal(&mut self, events: Vec<StreamEvent>) -> Result<()> {
        let start_time = Instant::now();
        let event_count = events.len();

        let result = match &mut self.backend_producer {
            #[cfg(feature = "kafka")]
            BackendProducer::Kafka(producer) => producer.publish_batch(events).await,
            #[cfg(feature = "nats")]
            BackendProducer::Nats(producer) => producer.publish_batch(events).await,
            #[cfg(feature = "redis")]
            BackendProducer::Redis(producer) => producer.publish_batch(events).await,
            #[cfg(feature = "kinesis")]
            BackendProducer::Kinesis(producer) => producer.publish_batch(events).await,
            #[cfg(feature = "pulsar")]
            BackendProducer::Pulsar(producer) => producer.publish_batch(events).await,
            #[cfg(feature = "rabbitmq")]
            BackendProducer::RabbitMQ(producer) => producer.publish_batch(events).await,
            BackendProducer::Memory(producer) => {
                for event in events {
                    producer.publish(event).await?;
                }
                Ok(())
            }
        };

        // Update stats
        let mut stats = self.stats.write().await;
        match &result {
            Ok(_) => {
                stats.events_published += event_count as u64;
                stats.batch_count += 1;
                let latency = start_time.elapsed().as_millis() as u64;
                stats.max_latency_ms = stats.max_latency_ms.max(latency);
                stats.avg_latency_ms = (stats.avg_latency_ms + latency as f64) / 2.0;
                stats.last_publish = Some(Utc::now());
            }
            Err(_) => {
                stats.events_failed += event_count as u64;
            }
        }

        debug!(
            "Published batch of {} events in {:?}",
            event_count,
            start_time.elapsed()
        );
        result
    }

    /// Flush any pending events and buffers
    pub async fn flush(&mut self) -> Result<()> {
        let _permit = self
            .flush_semaphore
            .acquire()
            .await
            .map_err(|_| anyhow!("Failed to acquire flush semaphore"))?;

        let start_time = Instant::now();

        // Flush any pending batch buffer
        if self.config.performance.enable_batching {
            let events = {
                let mut batch_buffer = self.batch_buffer.write().await;
                if !batch_buffer.is_empty() {
                    std::mem::take(&mut *batch_buffer)
                } else {
                    Vec::new()
                }
            };
            if !events.is_empty() {
                drop(_permit); // Release permit before mutable borrow
                self.publish_batch_internal(events).await?;
            }
        }

        // Flush backend-specific buffers
        let result = match &mut self.backend_producer {
            #[cfg(feature = "kafka")]
            BackendProducer::Kafka(producer) => producer.flush().await,
            #[cfg(feature = "nats")]
            BackendProducer::Nats(producer) => producer.flush().await,
            #[cfg(feature = "redis")]
            BackendProducer::Redis(producer) => producer.flush().await,
            #[cfg(feature = "kinesis")]
            BackendProducer::Kinesis(producer) => producer.flush().await,
            #[cfg(feature = "pulsar")]
            BackendProducer::Pulsar(producer) => producer.flush().await,
            #[cfg(feature = "rabbitmq")]
            BackendProducer::RabbitMQ(producer) => producer.flush().await,
            BackendProducer::Memory(producer) => producer.flush().await,
        };

        // Update stats
        if result.is_ok() {
            self.stats.write().await.flush_count += 1;
            self.last_flush = Instant::now();
            debug!("Flushed producer buffers in {:?}", start_time.elapsed());
        }

        result
    }

    /// Publish an RDF patch as a series of events
    pub async fn publish_patch(&mut self, patch: &RdfPatch) -> Result<()> {
        let events: Vec<StreamEvent> = patch
            .operations
            .iter()
            .filter_map(|op| {
                let metadata = EventMetadata {
                    event_id: Uuid::new_v4().to_string(),
                    timestamp: patch.timestamp,
                    source: "rdf_patch".to_string(),
                    user: None,
                    context: Some(patch.id.clone()),
                    caused_by: None,
                    version: "1.0".to_string(),
                    properties: HashMap::new(),
                    checksum: None,
                };

                match op {
                    PatchOperation::Add {
                        subject,
                        predicate,
                        object,
                    } => Some(StreamEvent::TripleAdded {
                        subject: subject.clone(),
                        predicate: predicate.clone(),
                        object: object.clone(),
                        graph: None,
                        metadata,
                    }),
                    PatchOperation::Delete {
                        subject,
                        predicate,
                        object,
                    } => Some(StreamEvent::TripleRemoved {
                        subject: subject.clone(),
                        predicate: predicate.clone(),
                        object: object.clone(),
                        graph: None,
                        metadata,
                    }),
                    PatchOperation::AddGraph { graph } => Some(StreamEvent::GraphCreated {
                        graph: graph.clone(),
                        metadata,
                    }),
                    PatchOperation::DeleteGraph { graph } => Some(StreamEvent::GraphDeleted {
                        graph: graph.clone(),
                        metadata,
                    }),
                    PatchOperation::AddPrefix { .. } => {
                        // Skip prefix operations for now
                        None
                    }
                    PatchOperation::DeletePrefix { .. } => {
                        // Skip prefix operations for now
                        None
                    }
                    PatchOperation::TransactionBegin { .. } => {
                        Some(StreamEvent::TransactionBegin {
                            transaction_id: patch
                                .transaction_id
                                .clone()
                                .unwrap_or_else(|| Uuid::new_v4().to_string()),
                            isolation_level: None,
                            metadata,
                        })
                    }
                    PatchOperation::TransactionCommit => Some(StreamEvent::TransactionCommit {
                        transaction_id: patch
                            .transaction_id
                            .clone()
                            .unwrap_or_else(|| Uuid::new_v4().to_string()),
                        metadata,
                    }),
                    PatchOperation::TransactionAbort => Some(StreamEvent::TransactionAbort {
                        transaction_id: patch
                            .transaction_id
                            .clone()
                            .unwrap_or_else(|| Uuid::new_v4().to_string()),
                        metadata,
                    }),
                    PatchOperation::Header { .. } => {
                        // Skip header operations for now
                        None
                    }
                }
            })
            .collect();

        self.publish_batch(events).await
    }

    /// Get producer statistics
    pub async fn get_stats(&self) -> ProducerStats {
        self.stats.read().await.clone()
    }

    /// Get producer health status
    pub async fn health_check(&self) -> bool {
        if let Some(cb) = &self.circuit_breaker {
            cb.is_healthy().await
        } else {
            true
        }
    }
}

/// Enhanced stream consumer for receiving RDF changes with backend support
pub struct StreamConsumer {
    _config: StreamConfig,
    backend_consumer: BackendConsumer,
    stats: Arc<RwLock<ConsumerStats>>,
    circuit_breaker: Option<circuit_breaker::SharedCircuitBreaker>,
    last_poll: Instant,
    _message_buffer: Arc<RwLock<Vec<StreamEvent>>>,
    consumer_group: Option<String>,
}

/// Backend-agnostic consumer wrapper
enum BackendConsumer {
    #[cfg(feature = "kafka")]
    Kafka(backend::kafka::KafkaConsumer),
    #[cfg(feature = "nats")]
    Nats(Box<backend::nats::NatsConsumer>),
    #[cfg(feature = "redis")]
    Redis(backend::redis::RedisConsumer),
    #[cfg(feature = "kinesis")]
    Kinesis(backend::kinesis::KinesisConsumer),
    #[cfg(feature = "pulsar")]
    Pulsar(Box<backend::pulsar::PulsarConsumer>),
    #[cfg(feature = "rabbitmq")]
    RabbitMQ(Box<backend::rabbitmq::RabbitMQConsumer>),
    Memory(MemoryConsumer),
}

/// Consumer statistics for monitoring
#[derive(Debug, Default, Clone)]
pub struct ConsumerStats {
    events_consumed: u64,
    events_failed: u64,
    _bytes_received: u64,
    avg_processing_time_ms: f64,
    max_processing_time_ms: u64,
    _consumer_lag: u64,
    circuit_breaker_trips: u64,
    last_message: Option<DateTime<Utc>>,
    backend_type: String,
    _batch_size: usize,
}

/// Memory-based consumer for testing and development
struct MemoryConsumer {
    backend: Box<dyn StreamBackend + Send + Sync>,
    topic: String,
    current_offset: u64,
    stats: ConsumerStats,
}

impl MemoryConsumer {
    fn _new() -> Self {
        Self {
            backend: Box::new(backend::memory::MemoryBackend::new()),
            topic: "oxirs-stream".to_string(),
            current_offset: 0,
            stats: ConsumerStats {
                backend_type: "memory".to_string(),
                ..Default::default()
            },
        }
    }

    fn with_topic(topic: String) -> Self {
        Self {
            backend: Box::new(backend::memory::MemoryBackend::new()),
            topic,
            current_offset: 0,
            stats: ConsumerStats {
                backend_type: "memory".to_string(),
                ..Default::default()
            },
        }
    }

    async fn consume(&mut self) -> Result<Option<StreamEvent>> {
        let start_time = Instant::now();

        // Use the backend implementation
        self.backend
            .as_mut()
            .connect()
            .await
            .map_err(|e| anyhow!("Backend connect failed: {}", e))?;

        // Try to receive events from the backend
        let topic_name = crate::types::TopicName::new(self.topic.clone());
        let events = self
            .backend
            .receive_events(
                &topic_name,
                None, // No consumer group for now
                crate::types::StreamPosition::Offset(self.current_offset),
                1, // Get one event at a time
            )
            .await
            .map_err(|e| anyhow!("Receive events failed: {}", e))?;

        if let Some((event, offset)) = events.first() {
            self.current_offset = offset.value() + 1;

            // Update stats
            self.stats.events_consumed += 1;
            let processing_time = start_time.elapsed().as_millis() as u64;
            self.stats.max_processing_time_ms =
                self.stats.max_processing_time_ms.max(processing_time);
            self.stats.avg_processing_time_ms =
                (self.stats.avg_processing_time_ms + processing_time as f64) / 2.0;
            self.stats.last_message = Some(Utc::now());

            debug!("Memory consumer: consumed event via backend");
            Ok(Some(event.clone()))
        } else {
            debug!("Memory consumer: no events available");
            Ok(None)
        }
    }

    fn _get_stats(&self) -> &ConsumerStats {
        &self.stats
    }

    /// Reset consumer position for testing
    fn reset(&mut self) {
        self.current_offset = 0;
    }
}

impl StreamConsumer {
    /// Create a new enhanced stream consumer with backend support
    pub async fn new(config: StreamConfig) -> Result<Self> {
        Self::new_with_group(config, None).await
    }

    /// Create a new stream consumer with a specific consumer group
    pub async fn new_with_group(
        config: StreamConfig,
        consumer_group: Option<String>,
    ) -> Result<Self> {
        // Initialize circuit breaker if enabled
        let circuit_breaker = if config.circuit_breaker.enabled {
            Some(circuit_breaker::new_shared_circuit_breaker(
                circuit_breaker::CircuitBreakerConfig {
                    enabled: config.circuit_breaker.enabled,
                    failure_threshold: config.circuit_breaker.failure_threshold,
                    success_threshold: config.circuit_breaker.success_threshold,
                    timeout: config.circuit_breaker.timeout,
                    half_open_max_calls: config.circuit_breaker.half_open_max_calls,
                    ..Default::default()
                },
            ))
        } else {
            None
        };

        // Initialize backend-specific consumer
        let backend_consumer = match &config.backend {
            #[cfg(feature = "kafka")]
            StreamBackendType::Kafka {
                brokers,
                security_protocol,
                sasl_config,
            } => {
                let stream_config = crate::StreamConfig {
                    backend: crate::StreamBackendType::Kafka {
                        brokers: brokers.clone(),
                        security_protocol: security_protocol.clone(),
                        sasl_config: sasl_config.clone(),
                    },
                    topic: config.topic.clone(),
                    batch_size: config.batch_size,
                    flush_interval_ms: config.flush_interval_ms,
                    max_connections: config.max_connections,
                    connection_timeout: config.connection_timeout,
                    enable_compression: config.enable_compression,
                    compression_type: config.compression_type.clone(),
                    retry_config: config.retry_config.clone(),
                    circuit_breaker: config.circuit_breaker.clone(),
                    security: config.security.clone(),
                    performance: config.performance.clone(),
                    monitoring: config.monitoring.clone(),
                };

                let mut consumer = backend::kafka::KafkaConsumer::new(stream_config)?;
                consumer.connect().await?;
                BackendConsumer::Kafka(consumer)
            }
            #[cfg(feature = "nats")]
            StreamBackendType::Nats {
                url,
                cluster_urls,
                jetstream_config,
            } => {
                let stream_config = crate::StreamConfig {
                    backend: crate::StreamBackendType::Nats {
                        url: url.clone(),
                        cluster_urls: cluster_urls.clone(),
                        jetstream_config: jetstream_config.clone(),
                    },
                    topic: config.topic.clone(),
                    batch_size: config.batch_size,
                    flush_interval_ms: config.flush_interval_ms,
                    max_connections: config.max_connections,
                    connection_timeout: config.connection_timeout,
                    enable_compression: config.enable_compression,
                    compression_type: config.compression_type.clone(),
                    retry_config: config.retry_config.clone(),
                    circuit_breaker: config.circuit_breaker.clone(),
                    security: config.security.clone(),
                    performance: config.performance.clone(),
                    monitoring: config.monitoring.clone(),
                };

                let mut consumer = backend::nats::NatsConsumer::new(stream_config)?;
                consumer.connect().await?;
                BackendConsumer::Nats(Box::new(consumer))
            }
            #[cfg(feature = "redis")]
            StreamBackendType::Redis {
                url,
                cluster_urls,
                pool_size,
            } => {
                let stream_config = crate::StreamConfig {
                    backend: crate::StreamBackendType::Redis {
                        url: url.clone(),
                        cluster_urls: cluster_urls.clone(),
                        pool_size: *pool_size,
                    },
                    topic: config.topic.clone(),
                    batch_size: config.batch_size,
                    flush_interval_ms: config.flush_interval_ms,
                    max_connections: config.max_connections,
                    connection_timeout: config.connection_timeout,
                    enable_compression: config.enable_compression,
                    compression_type: config.compression_type.clone(),
                    retry_config: config.retry_config.clone(),
                    circuit_breaker: config.circuit_breaker.clone(),
                    security: config.security.clone(),
                    performance: config.performance.clone(),
                    monitoring: config.monitoring.clone(),
                };

                let mut consumer = backend::redis::RedisConsumer::new(stream_config)?;
                consumer.connect().await?;
                BackendConsumer::Redis(consumer)
            }
            #[cfg(feature = "kinesis")]
            StreamBackendType::Kinesis {
                region,
                stream_name,
                credentials,
            } => {
                let stream_config = crate::StreamConfig {
                    backend: crate::StreamBackendType::Kinesis {
                        region: region.clone(),
                        stream_name: stream_name.clone(),
                        credentials: credentials.clone(),
                    },
                    topic: config.topic.clone(),
                    batch_size: config.batch_size,
                    flush_interval_ms: config.flush_interval_ms,
                    max_connections: config.max_connections,
                    connection_timeout: config.connection_timeout,
                    enable_compression: config.enable_compression,
                    compression_type: config.compression_type.clone(),
                    retry_config: config.retry_config.clone(),
                    circuit_breaker: config.circuit_breaker.clone(),
                    security: config.security.clone(),
                    performance: config.performance.clone(),
                    monitoring: config.monitoring.clone(),
                };

                let mut consumer = backend::kinesis::KinesisConsumer::new(stream_config)?;
                consumer.connect().await?;
                BackendConsumer::Kinesis(consumer)
            }
            #[cfg(feature = "pulsar")]
            StreamBackendType::Pulsar {
                service_url,
                auth_config,
            } => {
                let stream_config = crate::StreamConfig {
                    backend: crate::StreamBackendType::Pulsar {
                        service_url: service_url.clone(),
                        auth_config: auth_config.clone(),
                    },
                    topic: config.topic.clone(),
                    batch_size: config.batch_size,
                    flush_interval_ms: config.flush_interval_ms,
                    max_connections: config.max_connections,
                    connection_timeout: config.connection_timeout,
                    enable_compression: config.enable_compression,
                    compression_type: config.compression_type.clone(),
                    retry_config: config.retry_config.clone(),
                    circuit_breaker: config.circuit_breaker.clone(),
                    security: config.security.clone(),
                    performance: config.performance.clone(),
                    monitoring: config.monitoring.clone(),
                };

                let mut consumer = backend::pulsar::PulsarConsumer::new(stream_config)?;
                consumer.connect().await?;
                BackendConsumer::Pulsar(Box::new(consumer))
            }
            #[cfg(feature = "rabbitmq")]
            StreamBackendType::RabbitMQ {
                url,
                exchange,
                queue,
            } => {
                let stream_config = crate::StreamConfig {
                    backend: crate::StreamBackendType::RabbitMQ {
                        url: url.clone(),
                        exchange: exchange.clone(),
                        queue: queue.clone(),
                    },
                    topic: config.topic.clone(),
                    batch_size: config.batch_size,
                    flush_interval_ms: config.flush_interval_ms,
                    max_connections: config.max_connections,
                    connection_timeout: config.connection_timeout,
                    enable_compression: config.enable_compression,
                    compression_type: config.compression_type.clone(),
                    retry_config: config.retry_config.clone(),
                    circuit_breaker: config.circuit_breaker.clone(),
                    security: config.security.clone(),
                    performance: config.performance.clone(),
                    monitoring: config.monitoring.clone(),
                };

                let mut consumer = backend::rabbitmq::RabbitMQConsumer::new(stream_config)?;
                consumer.connect().await?;
                BackendConsumer::RabbitMQ(Box::new(consumer))
            }
            StreamBackendType::Memory {
                max_size: _,
                persistence: _,
            } => BackendConsumer::Memory(MemoryConsumer::with_topic(config.topic.clone())),
        };

        let stats = Arc::new(RwLock::new(ConsumerStats {
            backend_type: match backend_consumer {
                #[cfg(feature = "kafka")]
                BackendConsumer::Kafka(_) => "kafka".to_string(),
                #[cfg(feature = "nats")]
                BackendConsumer::Nats(_) => "nats".to_string(),
                #[cfg(feature = "redis")]
                BackendConsumer::Redis(_) => "redis".to_string(),
                #[cfg(feature = "kinesis")]
                BackendConsumer::Kinesis(_) => "kinesis".to_string(),
                #[cfg(feature = "pulsar")]
                BackendConsumer::Pulsar(_) => "pulsar".to_string(),
                #[cfg(feature = "rabbitmq")]
                BackendConsumer::RabbitMQ(_) => "rabbitmq".to_string(),
                BackendConsumer::Memory(_) => "memory".to_string(),
            },
            _batch_size: config.batch_size,
            ..Default::default()
        }));

        info!(
            "Created stream consumer with backend: {} and group: {:?}",
            stats.read().await.backend_type,
            consumer_group
        );

        Ok(Self {
            _config: config,
            backend_consumer,
            stats,
            circuit_breaker,
            last_poll: Instant::now(),
            _message_buffer: Arc::new(RwLock::new(Vec::new())),
            consumer_group,
        })
    }

    /// Consume stream events with circuit breaker protection
    pub async fn consume(&mut self) -> Result<Option<StreamEvent>> {
        let start_time = Instant::now();

        // Check circuit breaker if enabled
        if let Some(cb) = &self.circuit_breaker {
            if !cb.can_execute().await {
                self.stats.write().await.circuit_breaker_trips += 1;
                return Err(anyhow!("Circuit breaker is open - cannot consume events"));
            }
        }

        // Consume from backend
        let result = self.consume_single_event().await;

        // Update circuit breaker and stats
        match &result {
            Ok(Some(_)) => {
                if let Some(cb) = &self.circuit_breaker {
                    cb.record_success_with_duration(start_time.elapsed()).await;
                }

                let mut stats = self.stats.write().await;
                stats.events_consumed += 1;
                let processing_time = start_time.elapsed().as_millis() as u64;
                stats.max_processing_time_ms = stats.max_processing_time_ms.max(processing_time);
                stats.avg_processing_time_ms =
                    (stats.avg_processing_time_ms + processing_time as f64) / 2.0;
                stats.last_message = Some(Utc::now());
            }
            Ok(None) => {
                // No message available, not an error
                if let Some(cb) = &self.circuit_breaker {
                    cb.record_success_with_duration(start_time.elapsed()).await;
                }
            }
            Err(_) => {
                if let Some(cb) = &self.circuit_breaker {
                    cb.record_failure_with_type(circuit_breaker::FailureType::NetworkError)
                        .await;
                }

                self.stats.write().await.events_failed += 1;
            }
        }

        self.last_poll = Instant::now();
        result
    }

    /// Consume a single event from the backend
    async fn consume_single_event(&mut self) -> Result<Option<StreamEvent>> {
        match &mut self.backend_consumer {
            #[cfg(feature = "kafka")]
            BackendConsumer::Kafka(consumer) => consumer.consume().await,
            #[cfg(feature = "nats")]
            BackendConsumer::Nats(consumer) => consumer.consume().await,
            #[cfg(feature = "redis")]
            BackendConsumer::Redis(consumer) => consumer.consume().await,
            #[cfg(feature = "kinesis")]
            BackendConsumer::Kinesis(consumer) => consumer.consume().await,
            #[cfg(feature = "pulsar")]
            BackendConsumer::Pulsar(consumer) => consumer.consume().await,
            #[cfg(feature = "rabbitmq")]
            BackendConsumer::RabbitMQ(consumer) => consumer.consume().await,
            BackendConsumer::Memory(consumer) => consumer.consume().await,
        }
    }

    /// Consume multiple events as a batch
    pub async fn consume_batch(
        &mut self,
        max_events: usize,
        timeout: Duration,
    ) -> Result<Vec<StreamEvent>> {
        let mut events = Vec::new();
        let start_time = Instant::now();

        while events.len() < max_events && start_time.elapsed() < timeout {
            match tokio::time::timeout(Duration::from_millis(50), self.consume()).await {
                Ok(Ok(Some(event))) => events.push(event),
                Ok(Ok(None)) => continue,
                Ok(Err(e)) => return Err(e),
                Err(_) => break, // Timeout
            }
        }

        if !events.is_empty() {
            debug!(
                "Consumed batch of {} events in {:?}",
                events.len(),
                start_time.elapsed()
            );
        }

        Ok(events)
    }

    /// Start consuming events with a callback function
    pub async fn start_consuming<F>(&mut self, mut callback: F) -> Result<()>
    where
        F: FnMut(StreamEvent) -> Result<()> + Send,
    {
        info!("Starting stream consumer loop");

        loop {
            match self.consume().await {
                Ok(Some(event)) => {
                    if let Err(e) = callback(event) {
                        error!("Callback error: {}", e);
                        self.stats.write().await.events_failed += 1;
                    }
                }
                Ok(None) => {
                    // No message, wait a bit
                    tokio::time::sleep(Duration::from_millis(10)).await;
                }
                Err(e) => {
                    error!("Consumer error: {}", e);
                    tokio::time::sleep(Duration::from_millis(100)).await;
                }
            }
        }
    }

    /// Start consuming events with an async callback function
    pub async fn start_consuming_async<F, Fut>(&mut self, mut callback: F) -> Result<()>
    where
        F: FnMut(StreamEvent) -> Fut + Send,
        Fut: std::future::Future<Output = Result<()>> + Send,
    {
        info!("Starting async stream consumer loop");

        loop {
            match self.consume().await {
                Ok(Some(event)) => {
                    if let Err(e) = callback(event).await {
                        error!("Async callback error: {}", e);
                        self.stats.write().await.events_failed += 1;
                    }
                }
                Ok(None) => {
                    // No message, wait a bit
                    tokio::time::sleep(Duration::from_millis(10)).await;
                }
                Err(e) => {
                    error!("Consumer error: {}", e);
                    tokio::time::sleep(Duration::from_millis(100)).await;
                }
            }
        }
    }

    /// Get consumer statistics
    pub async fn get_stats(&self) -> ConsumerStats {
        self.stats.read().await.clone()
    }

    /// Get consumer health status
    pub async fn health_check(&self) -> bool {
        if let Some(cb) = &self.circuit_breaker {
            cb.is_healthy().await
        } else {
            true
        }
    }

    /// Get the consumer group name if any
    pub fn consumer_group(&self) -> Option<&String> {
        self.consumer_group.as_ref()
    }

    /// Reset consumer position (for testing with memory backend)
    pub async fn reset_position(&mut self) -> Result<()> {
        match &mut self.backend_consumer {
            BackendConsumer::Memory(consumer) => {
                consumer.reset();
                Ok(())
            }
            #[cfg(feature = "kafka")]
            BackendConsumer::Kafka(_) => {
                Err(anyhow!("Reset position not supported for Kafka backend"))
            }
            #[cfg(feature = "nats")]
            BackendConsumer::Nats(_) => {
                Err(anyhow!("Reset position not supported for NATS backend"))
            }
            #[cfg(feature = "redis")]
            BackendConsumer::Redis(_) => {
                Err(anyhow!("Reset position not supported for Redis backend"))
            }
            #[cfg(feature = "kinesis")]
            BackendConsumer::Kinesis(_) => {
                Err(anyhow!("Reset position not supported for Kinesis backend"))
            }
            #[cfg(feature = "pulsar")]
            BackendConsumer::Pulsar(_) => {
                Err(anyhow!("Reset position not supported for Pulsar backend"))
            }
            #[cfg(feature = "rabbitmq")]
            BackendConsumer::RabbitMQ(_) => {
                Err(anyhow!("Reset position not supported for RabbitMQ backend"))
            }
        }
    }

    /// Set test events for memory backend (for testing) - deprecated with backend implementation
    pub async fn set_test_events(&mut self, _events: Vec<StreamEvent>) -> Result<()> {
        // This method is deprecated with the new backend implementation
        // Events should be published through the producer instead
        Err(anyhow!("set_test_events is deprecated with backend implementation - use producer to publish events"))
    }
}

/// RDF patch operations with full protocol support
#[derive(Debug, Clone, Serialize, Deserialize)]
pub enum PatchOperation {
    /// Add a triple (A operation)
    Add {
        subject: String,
        predicate: String,
        object: String,
    },
    /// Delete a triple (D operation)
    Delete {
        subject: String,
        predicate: String,
        object: String,
    },
    /// Add a graph (GA operation)
    AddGraph { graph: String },
    /// Delete a graph (GD operation)
    DeleteGraph { graph: String },
    /// Add a prefix (PA operation)
    AddPrefix { prefix: String, namespace: String },
    /// Delete a prefix (PD operation)
    DeletePrefix { prefix: String },
    /// Transaction begin (TX operation)
    TransactionBegin { transaction_id: Option<String> },
    /// Transaction commit (TC operation)
    TransactionCommit,
    /// Transaction abort (TA operation)
    TransactionAbort,
    /// Header information (H operation)
    Header { key: String, value: String },
}

/// RDF patch for atomic updates with full protocol support
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct RdfPatch {
    pub operations: Vec<PatchOperation>,
    pub timestamp: chrono::DateTime<chrono::Utc>,
    pub id: String,
    /// Patch headers for metadata
    pub headers: HashMap<String, String>,
    /// Current transaction ID if in transaction
    pub transaction_id: Option<String>,
    /// Prefixes used in the patch
    pub prefixes: HashMap<String, String>,
}

impl RdfPatch {
    /// Create a new RDF patch
    pub fn new() -> Self {
        Self {
            operations: Vec::new(),
            timestamp: chrono::Utc::now(),
            id: uuid::Uuid::new_v4().to_string(),
            headers: HashMap::new(),
            transaction_id: None,
            prefixes: HashMap::new(),
        }
    }

    /// Add an operation to the patch
    pub fn add_operation(&mut self, operation: PatchOperation) {
        self.operations.push(operation);
    }

    /// Serialize patch to RDF Patch format
    pub fn to_rdf_patch_format(&self) -> Result<String> {
        let serializer = crate::patch::PatchSerializer::new()
            .with_pretty_print(true)
            .with_metadata(true);
        serializer.serialize(self)
    }

    /// Parse from RDF Patch format
    pub fn from_rdf_patch_format(input: &str) -> Result<Self> {
        let mut parser = crate::patch::PatchParser::new().with_strict_mode(false);
        parser.parse(input)
    }
}

impl Default for RdfPatch {
    fn default() -> Self {
        Self::new()
    }
}

// Default implementations for easier configuration
impl Default for StreamConfig {
    fn default() -> Self {
        Self {
            backend: StreamBackendType::Memory {
                max_size: Some(10000),
                persistence: false,
            },
            topic: "oxirs-stream".to_string(),
            batch_size: 100,
            flush_interval_ms: 100,
            max_connections: 10,
            connection_timeout: Duration::from_secs(30),
            enable_compression: false,
            compression_type: CompressionType::None,
            retry_config: RetryConfig::default(),
            circuit_breaker: CircuitBreakerConfig::default(),
            security: SecurityConfig::default(),
            performance: StreamPerformanceConfig::default(),
            monitoring: MonitoringConfig::default(),
        }
    }
}

impl Default for RetryConfig {
    fn default() -> Self {
        Self {
            max_retries: 3,
            initial_backoff: Duration::from_millis(100),
            max_backoff: Duration::from_secs(30),
            backoff_multiplier: 2.0,
            jitter: true,
        }
    }
}

impl Default for CircuitBreakerConfig {
    fn default() -> Self {
        Self {
            enabled: true,
            failure_threshold: 5,
            success_threshold: 3,
            timeout: Duration::from_secs(30),
            half_open_max_calls: 3,
        }
    }
}

impl Default for SecurityConfig {
    fn default() -> Self {
        Self {
            enable_tls: false,
            verify_certificates: true,
            client_cert_path: None,
            client_key_path: None,
            ca_cert_path: None,
            sasl_config: None,
        }
    }
}

impl Default for StreamPerformanceConfig {
    fn default() -> Self {
        Self {
            enable_batching: true,
            enable_pipelining: false,
            buffer_size: 8192,
            prefetch_count: 100,
            enable_zero_copy: false,
            enable_simd: false,
            parallel_processing: true,
            worker_threads: None,
        }
    }
}

impl Default for MonitoringConfig {
    fn default() -> Self {
        Self {
            enable_metrics: true,
            enable_tracing: true,
            metrics_interval: Duration::from_secs(60),
            health_check_interval: Duration::from_secs(30),
            enable_profiling: false,
            prometheus_endpoint: None,
            jaeger_endpoint: None,
            log_level: "info".to_string(),
        }
    }
}

/// Helper functions for creating common configurations
impl StreamConfig {
    /// Create a Redis configuration
    #[cfg(feature = "redis")]
    pub fn redis(url: String) -> Self {
        Self {
            backend: StreamBackendType::Redis {
                url,
                cluster_urls: None,
                pool_size: Some(10),
            },
            ..Default::default()
        }
    }

    /// Create a Kinesis configuration
    #[cfg(feature = "kinesis")]
    pub fn kinesis(region: String, stream_name: String) -> Self {
        Self {
            backend: StreamBackendType::Kinesis {
                region,
                stream_name,
                credentials: None,
            },
            ..Default::default()
        }
    }

    /// Create a memory configuration for testing
    pub fn memory() -> Self {
        Self {
            backend: StreamBackendType::Memory {
                max_size: Some(1000),
                persistence: false,
            },
            ..Default::default()
        }
    }

    /// Enable high-performance configuration
    pub fn high_performance(mut self) -> Self {
        self.performance.enable_batching = true;
        self.performance.enable_pipelining = true;
        self.performance.parallel_processing = true;
        self.performance.buffer_size = 65536;
        self.performance.prefetch_count = 1000;
        self.batch_size = 1000;
        self.flush_interval_ms = 10;
        self
    }

    /// Enable compression
    pub fn with_compression(mut self, compression_type: CompressionType) -> Self {
        self.enable_compression = true;
        self.compression_type = compression_type;
        self
    }

    /// Configure circuit breaker
    pub fn with_circuit_breaker(mut self, enabled: bool, failure_threshold: u32) -> Self {
        self.circuit_breaker.enabled = enabled;
        self.circuit_breaker.failure_threshold = failure_threshold;
        self
    }

    /// Create a development configuration with memory backend and debug settings
    pub fn development(topic: &str) -> Self {
        Self {
            backend: StreamBackendType::Memory {
                max_size: Some(10000),
                persistence: false,
            },
            topic: topic.to_string(),
            batch_size: 10,
            flush_interval_ms: 100,
            max_connections: 5,
            connection_timeout: Duration::from_secs(10),
            enable_compression: false,
            compression_type: CompressionType::None,
            retry_config: RetryConfig {
                max_retries: 3,
                initial_backoff: Duration::from_millis(100),
                max_backoff: Duration::from_secs(5),
                backoff_multiplier: 2.0,
                jitter: true,
            },
            circuit_breaker: CircuitBreakerConfig {
                enabled: false,
                failure_threshold: 5,
                success_threshold: 2,
                timeout: Duration::from_secs(60),
                half_open_max_calls: 10,
            },
            security: SecurityConfig::default(),
            performance: StreamPerformanceConfig::default(),
            monitoring: MonitoringConfig {
                enable_metrics: true,
                enable_tracing: false,
                metrics_interval: Duration::from_secs(5),
                health_check_interval: Duration::from_secs(30),
                enable_profiling: false,
                prometheus_endpoint: None,
                jaeger_endpoint: None,
                log_level: "debug".to_string(),
            },
        }
    }

    /// Create a production configuration with optimal performance settings
    pub fn production(topic: &str) -> Self {
        Self {
            backend: StreamBackendType::Memory {
                max_size: Some(100000),
                persistence: true,
            },
            topic: topic.to_string(),
            batch_size: 1000,
            flush_interval_ms: 10,
            max_connections: 50,
            connection_timeout: Duration::from_secs(30),
            enable_compression: true,
            compression_type: CompressionType::Zstd,
            retry_config: RetryConfig {
                max_retries: 5,
                initial_backoff: Duration::from_millis(200),
                max_backoff: Duration::from_secs(30),
                backoff_multiplier: 2.0,
                jitter: true,
            },
            circuit_breaker: CircuitBreakerConfig {
                enabled: true,
                failure_threshold: 10,
                success_threshold: 3,
                timeout: Duration::from_secs(30),
                half_open_max_calls: 5,
            },
            security: SecurityConfig::default(),
            performance: StreamPerformanceConfig {
                enable_batching: true,
                enable_pipelining: true,
                parallel_processing: true,
                buffer_size: 65536,
                prefetch_count: 1000,
                enable_zero_copy: true,
                enable_simd: true,
                worker_threads: None,
            },
            monitoring: MonitoringConfig {
                enable_metrics: true,
                enable_tracing: true,
                metrics_interval: Duration::from_secs(1),
                health_check_interval: Duration::from_secs(10),
                enable_profiling: true,
                prometheus_endpoint: None,
                jaeger_endpoint: None,
                log_level: "info".to_string(),
            },
        }
    }
}

/// Unified Stream interface that combines producer and consumer functionality
pub struct Stream {
    producer: StreamProducer,
    consumer: StreamConsumer,
}

impl Stream {
    /// Create a new unified stream instance
    pub async fn new(config: StreamConfig) -> Result<Self> {
        // Note: Commented out automatic clearing for performance tests
        // Clear memory events if using memory backend (important for testing)
        // if matches!(config.backend, StreamBackendType::Memory { .. }) {
        //     clear_memory_events().await;
        // }

        let producer = StreamProducer::new(config.clone()).await?;
        let consumer = StreamConsumer::new(config).await?;

        Ok(Self { producer, consumer })
    }

    /// Publish an event to the stream
    pub async fn publish(&mut self, event: StreamEvent) -> Result<()> {
        self.producer.publish(event).await
    }

    /// Consume an event from the stream
    pub async fn consume(&mut self) -> Result<Option<StreamEvent>> {
        self.consumer.consume().await
    }

    /// Flush any pending events
    pub async fn flush(&mut self) -> Result<()> {
        self.producer.flush().await
    }

    /// Get producer statistics
    pub async fn producer_stats(&self) -> ProducerStats {
        self.producer.get_stats().await
    }

    /// Get consumer statistics  
    pub async fn consumer_stats(&self) -> ConsumerStats {
        self.consumer.get_stats().await
    }

    /// Close the stream and clean up resources
    pub async fn close(&mut self) -> Result<()> {
        // Flush any pending events
        self.producer.flush().await?;

        // Close backend connections
        // Producer and consumer close operations would be handled by their Drop implementations
        debug!("Stream closed successfully");
        Ok(())
    }

    /// Perform a health check on the stream
    pub async fn health_check(&self) -> Result<bool> {
        // Basic health check - verify that the stream is operational
        // This is a simplified implementation
        Ok(true)
    }

    /// Begin a transaction (placeholder implementation)
    pub async fn begin_transaction(&mut self) -> Result<()> {
        // Placeholder implementation for transaction support
        debug!("Transaction begun (placeholder)");
        Ok(())
    }

    /// Commit a transaction (placeholder implementation)
    pub async fn commit_transaction(&mut self) -> Result<()> {
        // Placeholder implementation for transaction support
        debug!("Transaction committed (placeholder)");
        Ok(())
    }

    /// Rollback a transaction (placeholder implementation)
    pub async fn rollback_transaction(&mut self) -> Result<()> {
        // Placeholder implementation for transaction support
        debug!("Transaction rolled back (placeholder)");
        Ok(())
    }
}