varpulis_runtime/

tenant.rs

//! Multi-tenant isolation for SaaS deployment
//!
//! Provides tenant-scoped engine instances with resource limits and usage tracking.

use crate::context::ContextOrchestrator;
use crate::engine::Engine;
use crate::event::Event;
use crate::metrics::Metrics;
use crate::persistence::{StateStore, StoreError};
use serde::{Deserialize, Serialize};
use std::collections::HashMap;
use std::sync::atomic::{AtomicU64, Ordering};
use std::sync::Arc;
use std::time::Instant;
use tokio::sync::{mpsc, RwLock};
use tracing::{error, info, warn};
use uuid::Uuid;

/// Unique identifier for a tenant
#[derive(Debug, Clone, Hash, Eq, PartialEq, Serialize, Deserialize)]
pub struct TenantId(pub String);

impl TenantId {
    pub fn new(id: impl Into<String>) -> Self {
        Self(id.into())
    }

    pub fn generate() -> Self {
        Self(Uuid::new_v4().to_string())
    }

    pub fn as_str(&self) -> &str {
        &self.0
    }
}

impl std::fmt::Display for TenantId {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        write!(f, "{}", self.0)
    }
}

/// Resource limits for a tenant
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct TenantQuota {
    /// Maximum number of pipelines
    pub max_pipelines: usize,
    /// Maximum events per second
    pub max_events_per_second: u64,
    /// Maximum streams per pipeline
    pub max_streams_per_pipeline: usize,
}

impl Default for TenantQuota {
    fn default() -> Self {
        Self {
            max_pipelines: 10,
            max_events_per_second: 10_000,
            max_streams_per_pipeline: 50,
        }
    }
}

impl TenantQuota {
    /// Free tier limits
    pub const fn free() -> Self {
        Self {
            max_pipelines: 2,
            max_events_per_second: 100,
            max_streams_per_pipeline: 5,
        }
    }

    /// Pro tier limits
    pub const fn pro() -> Self {
        Self {
            max_pipelines: 20,
            max_events_per_second: 50_000,
            max_streams_per_pipeline: 100,
        }
    }

    /// Enterprise tier limits
    pub const fn enterprise() -> Self {
        Self {
            max_pipelines: 1000,
            max_events_per_second: 500_000,
            max_streams_per_pipeline: 500,
        }
    }
}

/// Usage statistics for a tenant
#[derive(Debug, Clone, Default)]
pub struct TenantUsage {
    /// Total events processed
    pub events_processed: u64,
    /// Events processed in current window (for rate limiting)
    pub events_in_window: u64,
    /// Window start time
    pub window_start: Option<Instant>,
    /// Number of active pipelines
    pub active_pipelines: usize,
    /// Total output events emitted
    pub output_events_emitted: u64,
}

impl TenantUsage {
    /// Record an event and check rate limit. Returns true if within quota.
    pub fn record_event(&mut self, max_eps: u64) -> bool {
        self.events_processed += 1;

        let now = Instant::now();
        match self.window_start {
            Some(start) if now.duration_since(start).as_secs() < 1 => {
                self.events_in_window += 1;
                if max_eps > 0 && self.events_in_window > max_eps {
                    return false;
                }
            }
            _ => {
                self.window_start = Some(now);
                self.events_in_window = 1;
            }
        }
        true
    }

    pub const fn record_output_event(&mut self) {
        self.output_events_emitted += 1;
    }
}

/// A pipeline deployed by a tenant
#[derive(Debug)]
pub struct Pipeline {
    /// Pipeline identifier
    pub id: String,
    /// Human-readable name
    pub name: String,
    /// The VPL source code
    pub source: String,
    /// The engine running this pipeline (shared with source ingestion loop)
    pub engine: Arc<tokio::sync::Mutex<Engine>>,
    /// Output event receiver for this pipeline
    pub output_rx: mpsc::Receiver<Event>,
    /// Broadcast sender for log streaming (SSE subscribers).
    /// Output events are forwarded here after being drained from output_rx.
    pub log_broadcast: tokio::sync::broadcast::Sender<Event>,
    /// When the pipeline was created
    pub created_at: Instant,
    /// Pipeline status
    pub status: PipelineStatus,
    /// Optional context orchestrator for multi-threaded execution
    pub orchestrator: Option<ContextOrchestrator>,
    /// Managed connector registry (keeps source connections alive)
    pub connector_registry: Option<crate::connector::ManagedConnectorRegistry>,
}

/// Pipeline status
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub enum PipelineStatus {
    Running,
    Stopped,
    Error(String),
}

impl std::fmt::Display for PipelineStatus {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        match self {
            Self::Running => write!(f, "running"),
            Self::Stopped => write!(f, "stopped"),
            Self::Error(msg) => write!(f, "error: {msg}"),
        }
    }
}

/// Errors from tenant operations
#[derive(Debug, thiserror::Error)]
pub enum TenantError {
    /// Tenant not found
    #[error("tenant not found: {0}")]
    NotFound(String),
    /// Pipeline not found
    #[error("pipeline not found: {0}")]
    PipelineNotFound(String),
    /// Quota exceeded
    #[error("quota exceeded: {0}")]
    QuotaExceeded(String),
    /// Rate limit exceeded
    #[error("rate limit exceeded")]
    RateLimitExceeded,
    /// Backpressure: queue depth exceeds configured maximum
    #[error("queue depth {current} exceeds maximum {max}")]
    BackpressureExceeded {
        /// Current queue depth
        current: u64,
        /// Configured maximum
        max: u64,
    },
    /// Parse error in VPL source
    #[error("parse error: {0}")]
    ParseError(#[from] varpulis_parser::ParseError),
    /// Engine error
    #[error("engine error: {0}")]
    EngineError(#[from] crate::engine::error::EngineError),
    /// Tenant already exists
    #[error("tenant already exists: {0}")]
    AlreadyExists(String),
}

/// A single tenant with its pipelines and quotas
#[derive(Debug)]
pub struct Tenant {
    /// Tenant identifier
    pub id: TenantId,
    /// Display name
    pub name: String,
    /// API key for authentication
    pub api_key: String,
    /// Resource quotas
    pub quota: TenantQuota,
    /// Usage statistics
    pub usage: TenantUsage,
    /// Active pipelines
    pub pipelines: HashMap<String, Pipeline>,
    /// When the tenant was created
    pub created_at: Instant,
    /// Optional WebSocket broadcast for output event relay
    pub(crate) ws_broadcast: Option<Arc<tokio::sync::broadcast::Sender<String>>>,
}

impl Tenant {
    pub fn new(id: TenantId, name: String, api_key: String, quota: TenantQuota) -> Self {
        Self {
            id,
            name,
            api_key,
            quota,
            usage: TenantUsage::default(),
            pipelines: HashMap::new(),
            created_at: Instant::now(),
            ws_broadcast: None,
        }
    }

    /// Deploy a new pipeline from VPL source code
    pub async fn deploy_pipeline(
        &mut self,
        name: String,
        source: String,
    ) -> Result<String, TenantError> {
        self.deploy_pipeline_with_metrics(name, source, None).await
    }

    pub async fn deploy_pipeline_with_metrics(
        &mut self,
        name: String,
        source: String,
        prometheus_metrics: Option<Metrics>,
    ) -> Result<String, TenantError> {
        // Check pipeline quota
        if self.pipelines.len() >= self.quota.max_pipelines {
            return Err(TenantError::QuotaExceeded(format!(
                "max pipelines ({}) reached",
                self.quota.max_pipelines
            )));
        }

        // Parse the VPL source
        let program = varpulis_parser::parse(&source)?;

        // Check streams quota
        let stream_count = program
            .statements
            .iter()
            .filter(|s| matches!(&s.node, varpulis_core::ast::Stmt::StreamDecl { .. }))
            .count();
        if stream_count > self.quota.max_streams_per_pipeline {
            return Err(TenantError::QuotaExceeded(format!(
                "pipeline has {} streams, max is {}",
                stream_count, self.quota.max_streams_per_pipeline
            )));
        }

        // Create a new engine
        let (output_tx, output_rx) = mpsc::channel(1000);
        let output_tx_for_ctx = output_tx.clone();
        let mut engine = Engine::new(output_tx);
        if let Some(m) = prometheus_metrics {
            engine = engine.with_metrics(m);
        }
        engine.load(&program)?;

        // Build context orchestrator if the program declares contexts
        let orchestrator = if engine.has_contexts() {
            // Context orchestrator connects sinks in each context thread
            match ContextOrchestrator::build(
                engine.context_map(),
                &program,
                output_tx_for_ctx,
                1000,
            ) {
                Ok(orch) => Some(orch),
                Err(e) => {
                    return Err(crate::engine::error::EngineError::Pipeline(format!(
                        "Failed to build context orchestrator: {e}"
                    ))
                    .into());
                }
            }
        } else {
            // No contexts - connect sinks on the main engine
            engine.connect_sinks().await?;
            None
        };

        // Start source connectors (MQTT/Kafka subscriptions)
        let bindings = engine.source_bindings().to_vec();
        let mut connector_registry = None;

        if !bindings.is_empty() {
            let (event_tx, event_rx) = mpsc::channel::<Event>(10_000);

            let mut registry = crate::connector::ManagedConnectorRegistry::from_configs(
                engine.connector_configs(),
            )
            .map_err(|e| {
                TenantError::EngineError(crate::engine::error::EngineError::Pipeline(format!(
                    "Registry build error: {e}"
                )))
            })?;

            for binding in &bindings {
                let config = engine.get_connector(&binding.connector_name).cloned();
                let Some(config) = config else {
                    return Err(crate::engine::error::EngineError::Pipeline(format!(
                        "Connector '{}' referenced in .from() but not declared",
                        binding.connector_name
                    ))
                    .into());
                };

                let topic = binding
                    .topic_override
                    .as_deref()
                    .or(config.topic.as_deref())
                    .unwrap_or("varpulis/events/#");

                info!(
                    "Starting {} source: {} topic={}",
                    config.connector_type, binding.connector_name, topic
                );

                registry
                    .start_source(
                        &binding.connector_name,
                        topic,
                        event_tx.clone(),
                        &binding.extra_params,
                    )
                    .await
                    .map_err(|e| -> TenantError {
                        crate::engine::error::EngineError::Pipeline(format!(
                            "Source start error: {e}"
                        ))
                        .into()
                    })?;

                // Create shared sinks for this connector
                let sink_keys = engine.sink_keys_for_connector(&binding.connector_name);
                for sink_key in &sink_keys {
                    let sink_topic = if let Some(t) =
                        sink_key.strip_prefix(&format!("{}::", binding.connector_name))
                    {
                        t.to_string()
                    } else {
                        config
                            .topic
                            .clone()
                            .unwrap_or_else(|| format!("{}-output", binding.connector_name))
                    };

                    let empty_params = std::collections::HashMap::new();
                    match registry.create_sink(&binding.connector_name, &sink_topic, &empty_params)
                    {
                        Ok(sink) => {
                            engine.inject_sink(sink_key, sink);
                        }
                        Err(e) => {
                            warn!("Failed to create sink for {}: {}", sink_key, e);
                        }
                    }
                }
            }

            connector_registry = Some(registry);

            // Wrap engine in Arc<Mutex> for sharing with source ingestion loop
            let engine = Arc::new(tokio::sync::Mutex::new(engine));
            let engine_for_source = Arc::clone(&engine);

            // Spawn source event ingestion loop
            tokio::spawn(async move {
                let mut event_rx = event_rx;
                while let Some(event) = event_rx.recv().await {
                    let mut eng = engine_for_source.lock().await;
                    if let Err(e) = eng.process(event).await {
                        warn!("Source event processing error: {}", e);
                    }
                }
                info!("Source ingestion loop ended");
            });

            let id = Uuid::new_v4().to_string();
            let (log_tx, _) = tokio::sync::broadcast::channel(256);

            // Spawn output event drain task: connector-sourced pipelines produce
            // output events via the engine's output_tx, but nobody calls
            // process_event() to drain them. This task continuously drains
            // output_rx → log_broadcast so WebSocket/SSE subscribers receive them.
            let log_tx_for_drain = log_tx.clone();
            let ws_tx_for_drain = self.ws_broadcast.clone();
            tokio::spawn(async move {
                let mut output_rx = output_rx;
                while let Some(ev) = output_rx.recv().await {
                    let _ = log_tx_for_drain.send(ev.clone());
                    // Forward to WebSocket relay channel if configured
                    if let Some(ref ws_tx) = ws_tx_for_drain {
                        let msg = serde_json::json!({
                            "type": "output_event",
                            "event_type": ev.event_type.to_string(),
                            "data": ev.data,
                            "timestamp": ev.timestamp.to_rfc3339(),
                        });
                        if let Ok(json) = serde_json::to_string(&msg) {
                            if ws_tx.send(json).is_err() {
                                tracing::debug!("No WS subscribers for drain output event");
                            }
                        }
                    }
                }
            });

            // The real output_rx was moved into the drain task above;
            // create a placeholder for the Pipeline struct.
            let (_drain_tx, placeholder_rx) = mpsc::channel(1);

            let pipeline = Pipeline {
                id: id.clone(),
                name,
                source,
                engine,
                output_rx: placeholder_rx,
                log_broadcast: log_tx,
                created_at: Instant::now(),
                status: PipelineStatus::Running,
                orchestrator,
                connector_registry,
            };

            self.pipelines.insert(id.clone(), pipeline);
            self.usage.active_pipelines = self.pipelines.len();

            return Ok(id);
        }

        // No source bindings — standard path without Arc<Mutex>
        let engine = Arc::new(tokio::sync::Mutex::new(engine));
        let id = Uuid::new_v4().to_string();
        let (log_tx, _) = tokio::sync::broadcast::channel(256);
        let pipeline = Pipeline {
            id: id.clone(),
            name,
            source,
            engine,
            output_rx,
            log_broadcast: log_tx,
            created_at: Instant::now(),
            status: PipelineStatus::Running,
            orchestrator,
            connector_registry,
        };

        self.pipelines.insert(id.clone(), pipeline);
        self.usage.active_pipelines = self.pipelines.len();

        Ok(id)
    }

    /// Remove a pipeline
    pub fn remove_pipeline(&mut self, pipeline_id: &str) -> Result<(), TenantError> {
        self.pipelines
            .remove(pipeline_id)
            .ok_or_else(|| TenantError::PipelineNotFound(pipeline_id.to_string()))?;
        self.usage.active_pipelines = self.pipelines.len();
        Ok(())
    }

    /// Process an event through a specific pipeline
    pub async fn process_event(
        &mut self,
        pipeline_id: &str,
        event: Event,
    ) -> Result<Vec<Event>, TenantError> {
        // Check rate limit
        if !self.usage.record_event(self.quota.max_events_per_second) {
            return Err(TenantError::RateLimitExceeded);
        }

        let pipeline = self
            .pipelines
            .get_mut(pipeline_id)
            .ok_or_else(|| TenantError::PipelineNotFound(pipeline_id.to_string()))?;

        if pipeline.status != PipelineStatus::Running {
            return Err(crate::engine::error::EngineError::Pipeline(format!(
                "pipeline is {}",
                pipeline.status
            ))
            .into());
        }

        if let Some(ref orchestrator) = pipeline.orchestrator {
            // Route through context orchestrator — try non-blocking first
            let shared_event = std::sync::Arc::new(event);
            match orchestrator.try_process(shared_event) {
                Ok(()) => {}
                Err(crate::context::DispatchError::ChannelFull(msg)) => {
                    // Extract the event from the ContextMessage and retry with await
                    if let crate::context::ContextMessage::Event(event) = msg {
                        orchestrator
                            .process(event)
                            .await
                            .map_err(|e| -> TenantError {
                                crate::engine::error::EngineError::Pipeline(e).into()
                            })?;
                    }
                }
                Err(crate::context::DispatchError::ChannelClosed(_)) => {
                    return Err(crate::engine::error::EngineError::Pipeline(
                        "Context channel closed".to_string(),
                    )
                    .into());
                }
            }
        } else {
            // Direct engine processing (no contexts, zero overhead)
            pipeline.engine.lock().await.process(event).await?;
        }

        // Drain output events from the channel and broadcast to log subscribers
        let mut output_events = Vec::new();
        while let Ok(ev) = pipeline.output_rx.try_recv() {
            // Best-effort broadcast to SSE log subscribers (ignore if no receivers)
            let _ = pipeline.log_broadcast.send(ev.clone());
            output_events.push(ev);
        }

        Ok(output_events)
    }

    /// Subscribe to a pipeline's output event stream for log streaming.
    /// Returns a broadcast receiver that yields events as they are produced.
    pub fn subscribe_pipeline_logs(
        &self,
        pipeline_id: &str,
    ) -> Result<tokio::sync::broadcast::Receiver<Event>, TenantError> {
        let pipeline = self
            .pipelines
            .get(pipeline_id)
            .ok_or_else(|| TenantError::PipelineNotFound(pipeline_id.to_string()))?;
        Ok(pipeline.log_broadcast.subscribe())
    }

    /// Create a checkpoint of a pipeline's engine state.
    pub async fn checkpoint_pipeline(
        &self,
        pipeline_id: &str,
    ) -> Result<crate::persistence::EngineCheckpoint, TenantError> {
        let pipeline = self
            .pipelines
            .get(pipeline_id)
            .ok_or_else(|| TenantError::PipelineNotFound(pipeline_id.to_string()))?;

        let engine = pipeline.engine.lock().await;
        Ok(engine.create_checkpoint())
    }

    /// Restore a pipeline's engine state from a checkpoint.
    pub async fn restore_pipeline(
        &mut self,
        pipeline_id: &str,
        checkpoint: &crate::persistence::EngineCheckpoint,
    ) -> Result<(), TenantError> {
        let pipeline = self
            .pipelines
            .get_mut(pipeline_id)
            .ok_or_else(|| TenantError::PipelineNotFound(pipeline_id.to_string()))?;

        let mut engine = pipeline.engine.lock().await;
        engine
            .restore_checkpoint(checkpoint)
            .map_err(crate::engine::error::EngineError::Store)?;
        Ok(())
    }

    /// Reload a pipeline with new VPL source
    pub async fn reload_pipeline(
        &mut self,
        pipeline_id: &str,
        source: String,
    ) -> Result<(), TenantError> {
        let program = varpulis_parser::parse(&source)?;

        let pipeline = self
            .pipelines
            .get_mut(pipeline_id)
            .ok_or_else(|| TenantError::PipelineNotFound(pipeline_id.to_string()))?;

        pipeline.engine.lock().await.reload(&program)?;
        pipeline.source = source;

        Ok(())
    }
}

/// Manages all tenants in the system
pub struct TenantManager {
    tenants: HashMap<TenantId, Tenant>,
    /// API key → TenantId lookup
    api_key_index: HashMap<String, TenantId>,
    /// Optional persistent state store
    store: Option<Arc<dyn StateStore>>,
    /// Shared Prometheus metrics (passed to engines on deploy)
    prometheus_metrics: Option<Metrics>,
    /// Maximum queue depth before rejecting events with backpressure (0 = unlimited)
    max_queue_depth: u64,
    /// Atomic counter of events currently being processed across all pipelines
    pending_events: Arc<AtomicU64>,
    /// Optional global output event broadcast for WebSocket relay.
    /// When set, all pipeline output events are forwarded here as JSON strings.
    ws_broadcast: Option<Arc<tokio::sync::broadcast::Sender<String>>>,
}

impl std::fmt::Debug for TenantManager {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("TenantManager")
            .field("tenant_count", &self.tenants.len())
            .field("max_queue_depth", &self.max_queue_depth)
            .field("pending_events", &self.pending_events)
            .field("has_store", &self.store.is_some())
            .field("has_ws_broadcast", &self.ws_broadcast.is_some())
            .finish_non_exhaustive()
    }
}

impl TenantManager {
    pub fn new() -> Self {
        Self {
            tenants: HashMap::new(),
            api_key_index: HashMap::new(),
            store: None,
            prometheus_metrics: None,
            max_queue_depth: 0,
            pending_events: Arc::new(AtomicU64::new(0)),
            ws_broadcast: None,
        }
    }

    /// Create a new tenant manager backed by a state store
    pub fn with_store(store: Arc<dyn StateStore>) -> Self {
        Self {
            tenants: HashMap::new(),
            api_key_index: HashMap::new(),
            store: Some(store),
            prometheus_metrics: None,
            max_queue_depth: 0,
            pending_events: Arc::new(AtomicU64::new(0)),
            ws_broadcast: None,
        }
    }

    /// Set Prometheus metrics to be shared with all engines
    pub fn set_prometheus_metrics(&mut self, metrics: Metrics) {
        self.prometheus_metrics = Some(metrics);
    }

    /// Set maximum queue depth for backpressure (0 = unlimited)
    pub const fn set_max_queue_depth(&mut self, max_depth: u64) {
        self.max_queue_depth = max_depth;
    }

    /// Set a WebSocket broadcast channel for relaying output events.
    /// When set, all connector-sourced pipeline output events are forwarded
    /// to this channel as serialized JSON strings.
    pub fn set_ws_broadcast(&mut self, tx: Arc<tokio::sync::broadcast::Sender<String>>) {
        for tenant in self.tenants.values_mut() {
            tenant.ws_broadcast = Some(Arc::clone(&tx));
        }
        self.ws_broadcast = Some(tx);
    }

    /// Get the maximum queue depth setting
    pub const fn max_queue_depth(&self) -> u64 {
        self.max_queue_depth
    }

    /// Get the current number of pending events being processed
    pub fn pending_event_count(&self) -> u64 {
        self.pending_events.load(Ordering::Relaxed)
    }

    /// Get a clone of the pending events counter (for sharing with API handlers)
    pub fn pending_events_counter(&self) -> Arc<AtomicU64> {
        Arc::clone(&self.pending_events)
    }

    /// Check backpressure: returns Err if queue depth exceeds maximum
    pub fn check_backpressure(&self) -> Result<(), TenantError> {
        if self.max_queue_depth == 0 {
            return Ok(());
        }
        let current = self.pending_events.load(Ordering::Relaxed);
        if current >= self.max_queue_depth {
            return Err(TenantError::BackpressureExceeded {
                current,
                max: self.max_queue_depth,
            });
        }
        Ok(())
    }

    /// Compute the queue pressure ratio (0.0 to 1.0+). Returns 0.0 if max_queue_depth is 0.
    pub fn queue_pressure_ratio(&self) -> f64 {
        if self.max_queue_depth == 0 {
            return 0.0;
        }
        self.pending_events.load(Ordering::Relaxed) as f64 / self.max_queue_depth as f64
    }

    /// Create a new tenant
    pub fn create_tenant(
        &mut self,
        name: String,
        api_key: String,
        quota: TenantQuota,
    ) -> Result<TenantId, TenantError> {
        if self.api_key_index.contains_key(&api_key) {
            return Err(TenantError::AlreadyExists(
                "API key already in use".to_string(),
            ));
        }

        let id = TenantId::generate();
        let mut tenant = Tenant::new(id.clone(), name, api_key.clone(), quota);
        tenant.ws_broadcast = self.ws_broadcast.clone();
        self.tenants.insert(id.clone(), tenant);
        self.api_key_index.insert(api_key, id.clone());
        self.persist_if_needed(&id);
        Ok(id)
    }

    /// Get a tenant by API key
    pub fn get_tenant_by_api_key(&self, api_key: &str) -> Option<&TenantId> {
        self.api_key_index.get(api_key)
    }

    /// Get a tenant by ID
    pub fn get_tenant(&self, id: &TenantId) -> Option<&Tenant> {
        self.tenants.get(id)
    }

    /// Get a mutable tenant by ID
    pub fn get_tenant_mut(&mut self, id: &TenantId) -> Option<&mut Tenant> {
        self.tenants.get_mut(id)
    }

    /// Deploy a pipeline on a tenant, passing through Prometheus metrics
    pub async fn deploy_pipeline_on_tenant(
        &mut self,
        tenant_id: &TenantId,
        name: String,
        source: String,
    ) -> Result<String, TenantError> {
        let metrics = self.prometheus_metrics.clone();
        let ws_broadcast = self.ws_broadcast.clone();
        let tenant = self
            .tenants
            .get_mut(tenant_id)
            .ok_or_else(|| TenantError::NotFound(tenant_id.to_string()))?;
        // Ensure tenant has the WS broadcast channel
        if tenant.ws_broadcast.is_none() {
            tenant.ws_broadcast = ws_broadcast;
        }
        tenant
            .deploy_pipeline_with_metrics(name, source, metrics)
            .await
    }

    /// Process an event through a pipeline with backpressure checking.
    ///
    /// Checks queue depth before processing, increments a pending counter during
    /// processing, and decrements it afterward. Returns `BackpressureExceeded`
    /// if the queue depth exceeds `max_queue_depth`.
    pub async fn process_event_with_backpressure(
        &mut self,
        tenant_id: &TenantId,
        pipeline_id: &str,
        event: Event,
    ) -> Result<Vec<Event>, TenantError> {
        self.check_backpressure()?;
        self.pending_events.fetch_add(1, Ordering::Relaxed);
        let tenant = self
            .tenants
            .get_mut(tenant_id)
            .ok_or_else(|| TenantError::NotFound(tenant_id.to_string()))?;
        let result = tenant.process_event(pipeline_id, event).await;
        self.pending_events.fetch_sub(1, Ordering::Relaxed);

        // Update pressure ratio metric
        if let Some(ref metrics) = self.prometheus_metrics {
            metrics
                .queue_pressure_ratio
                .with_label_values(&["_all"])
                .set(self.queue_pressure_ratio());
        }

        result
    }

    /// Remove a tenant and all its pipelines
    pub fn remove_tenant(&mut self, id: &TenantId) -> Result<(), TenantError> {
        let tenant = self
            .tenants
            .remove(id)
            .ok_or_else(|| TenantError::NotFound(id.to_string()))?;
        self.api_key_index.remove(&tenant.api_key);
        if let Some(ref store) = self.store {
            if let Err(e) = Self::delete_tenant_state(store.as_ref(), id) {
                warn!("Failed to delete persisted state for tenant {}: {}", id, e);
            }
        }
        Ok(())
    }

    /// List all tenants
    pub fn list_tenants(&self) -> Vec<&Tenant> {
        self.tenants.values().collect()
    }

    /// Get total number of tenants
    pub fn tenant_count(&self) -> usize {
        self.tenants.len()
    }

    /// Collect pipeline metrics across all tenants (pipeline_name, events_in, events_out).
    pub async fn collect_pipeline_metrics(&self) -> Vec<(String, u64, u64)> {
        let mut metrics = Vec::new();
        for tenant in self.tenants.values() {
            for pipeline in tenant.pipelines.values() {
                let engine = pipeline.engine.lock().await;
                let (events_in, events_out) = engine.event_counters();
                metrics.push((pipeline.name.clone(), events_in, events_out));
            }
        }
        metrics
    }

    /// Collect connector health across all tenants.
    ///
    /// Returns `(pipeline_name, connector_name, connector_type, health_report)` tuples.
    pub fn collect_connector_health(
        &self,
    ) -> Vec<(
        String,
        String,
        String,
        crate::connector::ConnectorHealthReport,
    )> {
        let mut results = Vec::new();
        for tenant in self.tenants.values() {
            for pipeline in tenant.pipelines.values() {
                if let Some(ref registry) = pipeline.connector_registry {
                    for (conn_name, conn_type, report) in registry.health_reports() {
                        results.push((
                            pipeline.name.clone(),
                            conn_name.to_string(),
                            conn_type.to_string(),
                            report,
                        ));
                    }
                }
            }
        }
        results
    }

    /// Persist a tenant's current state to the store (if configured)
    pub fn persist_if_needed(&self, tenant_id: &TenantId) {
        if let Some(ref store) = self.store {
            if let Some(tenant) = self.tenants.get(tenant_id) {
                if let Err(e) = Self::persist_tenant_to_store(store.as_ref(), tenant) {
                    warn!("Failed to persist tenant {}: {}", tenant_id, e);
                }
            }
        }
    }

    /// Recover all tenants from the state store
    ///
    /// Returns the number of tenants successfully recovered.
    pub fn recover(&mut self) -> Result<usize, StoreError> {
        let store = match &self.store {
            Some(s) => Arc::clone(s),
            None => return Ok(0),
        };

        // Load tenant index
        let index_data = match store.get("tenants:index")? {
            Some(data) => data,
            None => return Ok(0),
        };

        let tenant_ids: Vec<String> = serde_json::from_slice(&index_data)
            .map_err(|e| StoreError::SerializationError(e.to_string()))?;

        let mut recovered = 0;
        let mut failed = 0;

        for tid in &tenant_ids {
            let key = format!("tenant:{tid}");
            let data = match store.get(&key)? {
                Some(d) => d,
                None => {
                    warn!("Tenant {} listed in index but not found in store", tid);
                    failed += 1;
                    continue;
                }
            };

            let snapshot: TenantSnapshot = match serde_json::from_slice(&data) {
                Ok(s) => s,
                Err(e) => {
                    warn!("Failed to deserialize tenant {}: {}", tid, e);
                    failed += 1;
                    continue;
                }
            };

            match Self::restore_tenant_from_snapshot(snapshot) {
                Ok(tenant) => {
                    let tenant_id = tenant.id.clone();
                    self.api_key_index
                        .insert(tenant.api_key.clone(), tenant_id.clone());
                    let pipeline_count = tenant.pipelines.len();
                    self.tenants.insert(tenant_id.clone(), tenant);
                    info!(
                        "Recovered tenant {} with {} pipeline(s)",
                        tenant_id, pipeline_count
                    );
                    recovered += 1;
                }
                Err(e) => {
                    warn!("Failed to restore tenant {}: {}", tid, e);
                    failed += 1;
                }
            }
        }

        if failed > 0 {
            warn!(
                "Recovery complete: {} recovered, {} failed",
                recovered, failed
            );
        }

        Ok(recovered)
    }

    fn persist_tenant_to_store(store: &dyn StateStore, tenant: &Tenant) -> Result<(), StoreError> {
        let snapshot = tenant.snapshot();
        let data = serde_json::to_vec(&snapshot)
            .map_err(|e| StoreError::SerializationError(e.to_string()))?;
        let key = format!("tenant:{}", snapshot.id);
        store.put(&key, &data)?;

        // Update tenant index
        Self::update_tenant_index_add(store, &snapshot.id)?;

        store.flush()
    }

    fn delete_tenant_state(store: &dyn StateStore, id: &TenantId) -> Result<(), StoreError> {
        let key = format!("tenant:{}", id.0);
        store.delete(&key)?;

        Self::update_tenant_index_remove(store, &id.0)?;

        store.flush()
    }

    fn update_tenant_index_add(store: &dyn StateStore, tenant_id: &str) -> Result<(), StoreError> {
        let mut ids = Self::load_tenant_index(store)?;
        let id_str = tenant_id.to_string();
        if !ids.contains(&id_str) {
            ids.push(id_str);
        }
        let data =
            serde_json::to_vec(&ids).map_err(|e| StoreError::SerializationError(e.to_string()))?;
        store.put("tenants:index", &data)
    }

    fn update_tenant_index_remove(
        store: &dyn StateStore,
        tenant_id: &str,
    ) -> Result<(), StoreError> {
        let mut ids = Self::load_tenant_index(store)?;
        ids.retain(|id| id != tenant_id);
        let data =
            serde_json::to_vec(&ids).map_err(|e| StoreError::SerializationError(e.to_string()))?;
        store.put("tenants:index", &data)
    }

    fn load_tenant_index(store: &dyn StateStore) -> Result<Vec<String>, StoreError> {
        match store.get("tenants:index")? {
            Some(data) => serde_json::from_slice(&data)
                .map_err(|e| StoreError::SerializationError(e.to_string())),
            None => Ok(Vec::new()),
        }
    }

    fn restore_tenant_from_snapshot(snapshot: TenantSnapshot) -> Result<Tenant, TenantError> {
        let tenant_id = TenantId::new(&snapshot.id);

        let mut tenant = Tenant::new(tenant_id, snapshot.name, snapshot.api_key, snapshot.quota);
        tenant.usage.events_processed = snapshot.events_processed;
        tenant.usage.output_events_emitted = snapshot.output_events_emitted;
        tenant.usage.events_in_window = snapshot.events_in_window;

        for ps in snapshot.pipelines {
            match Self::restore_pipeline_from_snapshot(ps.clone()) {
                Ok(pipeline) => {
                    tenant.pipelines.insert(pipeline.id.clone(), pipeline);
                }
                Err(e) => {
                    warn!(
                        "Failed to restore pipeline '{}' ({}): {}",
                        ps.name, ps.id, e
                    );
                }
            }
        }

        tenant.usage.active_pipelines = tenant.pipelines.len();
        Ok(tenant)
    }

    fn restore_pipeline_from_snapshot(snapshot: PipelineSnapshot) -> Result<Pipeline, TenantError> {
        let program = varpulis_parser::parse(&snapshot.source)?;

        let (output_tx, output_rx) = mpsc::channel(1000);
        let mut engine = Engine::new(output_tx);
        engine.load(&program)?;

        let (log_tx, _) = tokio::sync::broadcast::channel(256);
        Ok(Pipeline {
            id: snapshot.id,
            name: snapshot.name,
            source: snapshot.source,
            engine: Arc::new(tokio::sync::Mutex::new(engine)),
            output_rx,
            log_broadcast: log_tx,
            created_at: Instant::now(),
            status: snapshot.status,
            orchestrator: None,
            connector_registry: None,
        })
    }
}

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

/// Thread-safe tenant manager for use in async server context
pub type SharedTenantManager = Arc<RwLock<TenantManager>>;

/// Create a new shared tenant manager
pub fn shared_tenant_manager() -> SharedTenantManager {
    Arc::new(RwLock::new(TenantManager::new()))
}

/// Create a shared tenant manager backed by a state store, recovering any persisted state
pub fn shared_tenant_manager_with_store(store: Arc<dyn StateStore>) -> SharedTenantManager {
    let mut mgr = TenantManager::with_store(store);
    match mgr.recover() {
        Ok(count) if count > 0 => {
            info!("Recovered {} tenant(s) from persistent state", count);
        }
        Ok(_) => {
            info!("No persisted tenant state found, starting fresh");
        }
        Err(e) => {
            error!("Failed to recover tenant state: {}", e);
        }
    }
    Arc::new(RwLock::new(mgr))
}

// =============================================================================
// Snapshot Types for Persistence
// =============================================================================

/// Serializable snapshot of a tenant's state
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct TenantSnapshot {
    pub id: String,
    pub name: String,
    pub api_key: String,
    pub quota: TenantQuota,
    pub events_processed: u64,
    #[serde(alias = "alerts_generated")]
    pub output_events_emitted: u64,
    pub pipelines: Vec<PipelineSnapshot>,
    #[serde(default)]
    pub created_at_ms: Option<i64>,
    #[serde(default)]
    pub events_in_window: u64,
}

/// Serializable snapshot of a pipeline
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct PipelineSnapshot {
    pub id: String,
    pub name: String,
    pub source: String,
    pub status: PipelineStatus,
}

impl Pipeline {
    /// Create a serializable snapshot of this pipeline
    pub fn snapshot(&self) -> PipelineSnapshot {
        PipelineSnapshot {
            id: self.id.clone(),
            name: self.name.clone(),
            source: self.source.clone(),
            status: self.status.clone(),
        }
    }
}

impl Tenant {
    /// Create a serializable snapshot of this tenant and all its pipelines
    pub fn snapshot(&self) -> TenantSnapshot {
        TenantSnapshot {
            id: self.id.0.clone(),
            name: self.name.clone(),
            api_key: self.api_key.clone(),
            quota: self.quota.clone(),
            events_processed: self.usage.events_processed,
            output_events_emitted: self.usage.output_events_emitted,
            pipelines: self.pipelines.values().map(|p| p.snapshot()).collect(),
            created_at_ms: Some(chrono::Utc::now().timestamp_millis()),
            events_in_window: self.usage.events_in_window,
        }
    }
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn test_tenant_id_generate() {
        let id1 = TenantId::generate();
        let id2 = TenantId::generate();
        assert_ne!(id1, id2);
    }

    #[test]
    fn test_tenant_id_display() {
        let id = TenantId::new("test-123");
        assert_eq!(format!("{id}"), "test-123");
        assert_eq!(id.as_str(), "test-123");
    }

    #[test]
    fn test_tenant_quota_tiers() {
        let free = TenantQuota::free();
        let pro = TenantQuota::pro();
        let enterprise = TenantQuota::enterprise();

        assert!(free.max_pipelines < pro.max_pipelines);
        assert!(pro.max_pipelines < enterprise.max_pipelines);
        assert!(free.max_events_per_second < pro.max_events_per_second);
        assert!(pro.max_events_per_second < enterprise.max_events_per_second);
    }

    #[test]
    fn test_usage_record_event() {
        let mut usage = TenantUsage::default();
        assert!(usage.record_event(100));
        assert_eq!(usage.events_processed, 1);
        assert_eq!(usage.events_in_window, 1);
    }

    #[test]
    fn test_usage_rate_limit() {
        let mut usage = TenantUsage::default();
        // Allow 2 events per second
        assert!(usage.record_event(2));
        assert!(usage.record_event(2));
        // Third should be rejected
        assert!(!usage.record_event(2));
    }

    #[test]
    fn test_usage_no_rate_limit() {
        let mut usage = TenantUsage::default();
        // 0 means disabled
        for _ in 0..1000 {
            assert!(usage.record_event(0));
        }
    }

    #[test]
    fn test_tenant_manager_create() {
        let mut mgr = TenantManager::new();
        let id = mgr
            .create_tenant("Test Corp".into(), "key-123".into(), TenantQuota::free())
            .unwrap();

        assert_eq!(mgr.tenant_count(), 1);
        assert!(mgr.get_tenant(&id).is_some());
        assert_eq!(mgr.get_tenant(&id).unwrap().name, "Test Corp");
    }

    #[test]
    fn test_tenant_manager_api_key_lookup() {
        let mut mgr = TenantManager::new();
        let id = mgr
            .create_tenant("Test".into(), "my-key".into(), TenantQuota::default())
            .unwrap();

        let found = mgr.get_tenant_by_api_key("my-key");
        assert_eq!(found, Some(&id));

        assert!(mgr.get_tenant_by_api_key("wrong-key").is_none());
    }

    #[test]
    fn test_tenant_manager_duplicate_api_key() {
        let mut mgr = TenantManager::new();
        mgr.create_tenant("A".into(), "key-1".into(), TenantQuota::default())
            .unwrap();
        let result = mgr.create_tenant("B".into(), "key-1".into(), TenantQuota::default());
        assert!(result.is_err());
    }

    #[test]
    fn test_tenant_manager_remove() {
        let mut mgr = TenantManager::new();
        let id = mgr
            .create_tenant("Test".into(), "key-1".into(), TenantQuota::default())
            .unwrap();

        mgr.remove_tenant(&id).unwrap();
        assert_eq!(mgr.tenant_count(), 0);
        assert!(mgr.get_tenant_by_api_key("key-1").is_none());
    }

    #[tokio::test]
    async fn test_tenant_deploy_pipeline() {
        let mut mgr = TenantManager::new();
        let id = mgr
            .create_tenant("Test".into(), "key-1".into(), TenantQuota::default())
            .unwrap();

        let tenant = mgr.get_tenant_mut(&id).unwrap();
        let vpl = r"
            stream Alerts = SensorReading
                .where(temperature > 100)
        ";
        let pipeline_id = tenant
            .deploy_pipeline("My Pipeline".into(), vpl.into())
            .await
            .unwrap();
        assert_eq!(tenant.pipelines.len(), 1);
        assert_eq!(tenant.usage.active_pipelines, 1);
        assert_eq!(
            tenant.pipelines[&pipeline_id].status,
            PipelineStatus::Running
        );
    }

    #[tokio::test]
    async fn test_tenant_pipeline_quota() {
        let mut mgr = TenantManager::new();
        let quota = TenantQuota {
            max_pipelines: 1,
            max_events_per_second: 100,
            max_streams_per_pipeline: 50,
        };
        let id = mgr
            .create_tenant("Test".into(), "key-1".into(), quota)
            .unwrap();

        let tenant = mgr.get_tenant_mut(&id).unwrap();
        let vpl = "stream A = SensorReading .where(x > 1)";
        tenant
            .deploy_pipeline("P1".into(), vpl.into())
            .await
            .unwrap();

        // Second should fail
        let result = tenant.deploy_pipeline("P2".into(), vpl.into()).await;
        assert!(result.is_err());
    }

    #[tokio::test]
    async fn test_tenant_remove_pipeline() {
        let mut mgr = TenantManager::new();
        let id = mgr
            .create_tenant("Test".into(), "key-1".into(), TenantQuota::default())
            .unwrap();

        let tenant = mgr.get_tenant_mut(&id).unwrap();
        let vpl = "stream A = SensorReading .where(x > 1)";
        let pid = tenant
            .deploy_pipeline("P1".into(), vpl.into())
            .await
            .unwrap();

        tenant.remove_pipeline(&pid).unwrap();
        assert_eq!(tenant.pipelines.len(), 0);
        assert_eq!(tenant.usage.active_pipelines, 0);
    }

    #[tokio::test]
    async fn test_tenant_parse_error() {
        let mut mgr = TenantManager::new();
        let id = mgr
            .create_tenant("Test".into(), "key-1".into(), TenantQuota::default())
            .unwrap();

        let tenant = mgr.get_tenant_mut(&id).unwrap();
        let result = tenant
            .deploy_pipeline("Bad".into(), "this is not valid VPL {{{{".into())
            .await;
        assert!(result.is_err());
    }

    #[tokio::test]
    async fn test_tenant_process_event() {
        let mut mgr = TenantManager::new();
        let id = mgr
            .create_tenant("Test".into(), "key-1".into(), TenantQuota::default())
            .unwrap();

        let tenant = mgr.get_tenant_mut(&id).unwrap();
        let vpl = "stream A = SensorReading .where(temperature > 100)";
        let pid = tenant
            .deploy_pipeline("P1".into(), vpl.into())
            .await
            .unwrap();

        let event = Event::new("SensorReading").with_field("temperature", 150.0);
        tenant.process_event(&pid, event).await.unwrap();
        assert_eq!(tenant.usage.events_processed, 1);
    }

    #[tokio::test]
    async fn test_tenant_rate_limit_on_process() {
        let mut mgr = TenantManager::new();
        let quota = TenantQuota {
            max_pipelines: 10,
            max_events_per_second: 2,
            max_streams_per_pipeline: 50,
        };
        let id = mgr
            .create_tenant("Test".into(), "key-1".into(), quota)
            .unwrap();

        let tenant = mgr.get_tenant_mut(&id).unwrap();
        let vpl = "stream A = SensorReading .where(x > 1)";
        let pid = tenant
            .deploy_pipeline("P1".into(), vpl.into())
            .await
            .unwrap();

        let event = Event::new("SensorReading").with_field("x", 5);
        tenant.process_event(&pid, event.clone()).await.unwrap();
        tenant.process_event(&pid, event.clone()).await.unwrap();

        // Third should hit rate limit
        let result = tenant.process_event(&pid, event).await;
        assert!(result.is_err());
    }

    #[tokio::test]
    async fn test_tenant_reload_pipeline() {
        let mut mgr = TenantManager::new();
        let id = mgr
            .create_tenant("Test".into(), "key-1".into(), TenantQuota::default())
            .unwrap();

        let tenant = mgr.get_tenant_mut(&id).unwrap();
        let vpl1 = "stream A = SensorReading .where(x > 1)";
        let pid = tenant
            .deploy_pipeline("P1".into(), vpl1.into())
            .await
            .unwrap();

        let vpl2 = "stream B = SensorReading .where(x > 50)";
        tenant.reload_pipeline(&pid, vpl2.into()).await.unwrap();

        assert_eq!(tenant.pipelines[&pid].source, vpl2);
    }

    #[test]
    fn test_pipeline_status_display() {
        assert_eq!(format!("{}", PipelineStatus::Running), "running");
        assert_eq!(format!("{}", PipelineStatus::Stopped), "stopped");
        assert_eq!(
            format!("{}", PipelineStatus::Error("oops".into())),
            "error: oops"
        );
    }

    #[test]
    fn test_tenant_error_display() {
        assert!(format!("{}", TenantError::NotFound("t1".into())).contains("t1"));
        assert!(format!("{}", TenantError::RateLimitExceeded).contains("rate limit"));
        let engine_err = crate::engine::error::EngineError::Compilation("bad".into());
        assert!(format!("{}", TenantError::EngineError(engine_err)).contains("bad"));
    }

    #[test]
    fn test_shared_tenant_manager() {
        let mgr = shared_tenant_manager();
        assert!(Arc::strong_count(&mgr) == 1);
    }

    #[test]
    fn test_tenant_list() {
        let mut mgr = TenantManager::new();
        mgr.create_tenant("A".into(), "key-a".into(), TenantQuota::default())
            .unwrap();
        mgr.create_tenant("B".into(), "key-b".into(), TenantQuota::default())
            .unwrap();
        assert_eq!(mgr.list_tenants().len(), 2);
    }

    #[test]
    fn test_tenant_manager_default() {
        let mgr = TenantManager::default();
        assert_eq!(mgr.tenant_count(), 0);
    }

    #[tokio::test]
    async fn test_tenant_snapshot_roundtrip() {
        let mut mgr = TenantManager::new();
        let id = mgr
            .create_tenant("Snap Corp".into(), "snap-key".into(), TenantQuota::pro())
            .unwrap();

        let tenant = mgr.get_tenant_mut(&id).unwrap();
        let vpl = "stream A = SensorReading .where(x > 1)";
        tenant
            .deploy_pipeline("Pipeline1".into(), vpl.into())
            .await
            .unwrap();
        tenant.usage.events_processed = 42;
        tenant.usage.output_events_emitted = 7;

        let snapshot = tenant.snapshot();
        let json = serde_json::to_vec(&snapshot).unwrap();
        let restored: TenantSnapshot = serde_json::from_slice(&json).unwrap();

        assert_eq!(restored.id, id.0);
        assert_eq!(restored.name, "Snap Corp");
        assert_eq!(restored.api_key, "snap-key");
        assert_eq!(restored.events_processed, 42);
        assert_eq!(restored.output_events_emitted, 7);
        assert_eq!(restored.pipelines.len(), 1);
        assert_eq!(restored.pipelines[0].name, "Pipeline1");
        assert_eq!(restored.pipelines[0].source, vpl);
        assert_eq!(restored.pipelines[0].status, PipelineStatus::Running);
        assert_eq!(
            restored.quota.max_pipelines,
            TenantQuota::pro().max_pipelines
        );
    }

    #[tokio::test]
    async fn test_tenant_manager_persistence_and_recovery() {
        use crate::persistence::MemoryStore;

        let store: Arc<dyn StateStore> = Arc::new(MemoryStore::new());
        let vpl = "stream A = SensorReading .where(x > 1)";

        // Create tenants and pipelines with persistence
        let (tenant_id, pipeline_name) = {
            let mut mgr = TenantManager::with_store(Arc::clone(&store));
            let id = mgr
                .create_tenant(
                    "Persisted Corp".into(),
                    "persist-key".into(),
                    TenantQuota::default(),
                )
                .unwrap();

            let tenant = mgr.get_tenant_mut(&id).unwrap();
            tenant
                .deploy_pipeline("Persistent Pipeline".into(), vpl.into())
                .await
                .unwrap();
            tenant.usage.events_processed = 100;
            tenant.usage.output_events_emitted = 5;
            mgr.persist_if_needed(&id);

            (id.0.clone(), "Persistent Pipeline".to_string())
        };

        // Recover into a new manager
        let mut mgr2 = TenantManager::with_store(Arc::clone(&store));
        let recovered = mgr2.recover().unwrap();
        assert_eq!(recovered, 1);
        assert_eq!(mgr2.tenant_count(), 1);

        // Verify tenant data
        let tid = TenantId::new(&tenant_id);
        let tenant = mgr2.get_tenant(&tid).unwrap();
        assert_eq!(tenant.name, "Persisted Corp");
        assert_eq!(tenant.api_key, "persist-key");
        assert_eq!(tenant.usage.events_processed, 100);
        assert_eq!(tenant.usage.output_events_emitted, 5);
        assert_eq!(tenant.pipelines.len(), 1);

        let pipeline = tenant.pipelines.values().next().unwrap();
        assert_eq!(pipeline.name, pipeline_name);
        assert_eq!(pipeline.source, vpl);
        assert_eq!(pipeline.status, PipelineStatus::Running);

        // Verify API key index was rebuilt
        assert_eq!(mgr2.get_tenant_by_api_key("persist-key"), Some(&tid));
    }

    #[tokio::test]
    async fn test_persistence_survives_restart() {
        use crate::persistence::MemoryStore;

        let store: Arc<dyn StateStore> = Arc::new(MemoryStore::new());
        let vpl = "stream A = SensorReading .where(x > 1)";

        // Phase 1: Create data
        {
            let mut mgr = TenantManager::with_store(Arc::clone(&store));
            let id1 = mgr
                .create_tenant("Tenant A".into(), "key-a".into(), TenantQuota::free())
                .unwrap();
            let id2 = mgr
                .create_tenant("Tenant B".into(), "key-b".into(), TenantQuota::pro())
                .unwrap();

            let tenant_a = mgr.get_tenant_mut(&id1).unwrap();
            tenant_a
                .deploy_pipeline("P1".into(), vpl.into())
                .await
                .unwrap();
            mgr.persist_if_needed(&id1);

            let tenant_b = mgr.get_tenant_mut(&id2).unwrap();
            tenant_b
                .deploy_pipeline("P2".into(), vpl.into())
                .await
                .unwrap();
            tenant_b
                .deploy_pipeline("P3".into(), vpl.into())
                .await
                .unwrap();
            mgr.persist_if_needed(&id2);
        }

        // Phase 2: Recover (simulating restart)
        {
            let mut mgr = TenantManager::with_store(Arc::clone(&store));
            let recovered = mgr.recover().unwrap();
            assert_eq!(recovered, 2);
            assert_eq!(mgr.tenant_count(), 2);

            // Verify tenant A
            let tid_a = mgr.get_tenant_by_api_key("key-a").unwrap().clone();
            let tenant_a = mgr.get_tenant(&tid_a).unwrap();
            assert_eq!(tenant_a.name, "Tenant A");
            assert_eq!(tenant_a.pipelines.len(), 1);

            // Verify tenant B
            let tid_b = mgr.get_tenant_by_api_key("key-b").unwrap().clone();
            let tenant_b = mgr.get_tenant(&tid_b).unwrap();
            assert_eq!(tenant_b.name, "Tenant B");
            assert_eq!(tenant_b.pipelines.len(), 2);
        }
    }

    #[test]
    fn test_tenant_snapshot_created_at_and_window() {
        let mut mgr = TenantManager::new();
        let id = mgr
            .create_tenant(
                "Window Corp".into(),
                "window-key".into(),
                TenantQuota::default(),
            )
            .unwrap();

        let tenant = mgr.get_tenant_mut(&id).unwrap();
        tenant.usage.events_in_window = 42;
        tenant.usage.events_processed = 100;

        let snapshot = tenant.snapshot();

        // created_at_ms should be populated
        assert!(snapshot.created_at_ms.is_some());
        let ts = snapshot.created_at_ms.unwrap();
        // Should be a reasonable recent timestamp (after 2024-01-01)
        assert!(ts > 1_704_067_200_000);

        // events_in_window should be captured
        assert_eq!(snapshot.events_in_window, 42);

        // Round-trip through JSON
        let json = serde_json::to_vec(&snapshot).unwrap();
        let restored: TenantSnapshot = serde_json::from_slice(&json).unwrap();
        assert_eq!(restored.events_in_window, 42);
        assert_eq!(restored.created_at_ms, snapshot.created_at_ms);

        // Restore tenant and verify events_in_window is carried over
        let restored_tenant = TenantManager::restore_tenant_from_snapshot(restored).unwrap();
        assert_eq!(restored_tenant.usage.events_in_window, 42);
        assert_eq!(restored_tenant.usage.events_processed, 100);
    }

    #[test]
    fn test_tenant_snapshot_backwards_compat() {
        // Simulate old JSON without the new fields
        let old_json = r#"{
            "id": "compat-tenant",
            "name": "Compat Corp",
            "api_key": "compat-key",
            "quota": {
                "max_pipelines": 10,
                "max_events_per_second": 10000,
                "max_streams_per_pipeline": 50
            },
            "events_processed": 55,
            "alerts_generated": 3,
            "pipelines": []
        }"#;

        let snapshot: TenantSnapshot = serde_json::from_str(old_json).unwrap();
        assert_eq!(snapshot.id, "compat-tenant");
        assert_eq!(snapshot.events_processed, 55);
        // New fields should default
        assert_eq!(snapshot.created_at_ms, None);
        assert_eq!(snapshot.events_in_window, 0);

        // Should restore fine
        let tenant = TenantManager::restore_tenant_from_snapshot(snapshot).unwrap();
        assert_eq!(tenant.name, "Compat Corp");
        assert_eq!(tenant.usage.events_processed, 55);
        assert_eq!(tenant.usage.events_in_window, 0);
    }

    #[test]
    fn test_recovery_with_invalid_vpl() {
        use crate::persistence::MemoryStore;

        let store: Arc<dyn StateStore> = Arc::new(MemoryStore::new());

        // Manually store a tenant with invalid VPL
        let snapshot = TenantSnapshot {
            id: "bad-tenant".into(),
            name: "Bad Corp".into(),
            api_key: "bad-key".into(),
            quota: TenantQuota::default(),
            events_processed: 0,
            output_events_emitted: 0,
            pipelines: vec![PipelineSnapshot {
                id: "bad-pipeline".into(),
                name: "Bad Pipeline".into(),
                source: "this is not valid VPL {{{{".into(),
                status: PipelineStatus::Running,
            }],
            created_at_ms: None,
            events_in_window: 0,
        };

        let data = serde_json::to_vec(&snapshot).unwrap();
        store.put("tenant:bad-tenant", &data).unwrap();
        let index = serde_json::to_vec(&vec!["bad-tenant"]).unwrap();
        store.put("tenants:index", &index).unwrap();

        // Recovery should succeed (tenant recovered, bad pipeline skipped)
        let mut mgr = TenantManager::with_store(Arc::clone(&store));
        let recovered = mgr.recover().unwrap();
        assert_eq!(recovered, 1);

        let tid = TenantId::new("bad-tenant");
        let tenant = mgr.get_tenant(&tid).unwrap();
        assert_eq!(tenant.name, "Bad Corp");
        // The invalid pipeline should have been skipped
        assert_eq!(tenant.pipelines.len(), 0);
    }

    #[test]
    fn test_backpressure_disabled_by_default() {
        let mgr = TenantManager::new();
        assert_eq!(mgr.max_queue_depth(), 0);
        assert!(mgr.check_backpressure().is_ok());
    }

    #[test]
    fn test_backpressure_set_max_queue_depth() {
        let mut mgr = TenantManager::new();
        mgr.set_max_queue_depth(100);
        assert_eq!(mgr.max_queue_depth(), 100);
        // No pending events, should pass
        assert!(mgr.check_backpressure().is_ok());
    }

    #[test]
    fn test_backpressure_exceeded() {
        let mut mgr = TenantManager::new();
        mgr.set_max_queue_depth(10);

        // Simulate 10 pending events
        mgr.pending_events.store(10, Ordering::Relaxed);
        let result = mgr.check_backpressure();
        assert!(result.is_err());

        if let Err(TenantError::BackpressureExceeded { current, max }) = result {
            assert_eq!(current, 10);
            assert_eq!(max, 10);
        } else {
            panic!("Expected BackpressureExceeded error");
        }
    }

    #[test]
    fn test_backpressure_not_exceeded() {
        let mut mgr = TenantManager::new();
        mgr.set_max_queue_depth(10);
        mgr.pending_events.store(9, Ordering::Relaxed);
        assert!(mgr.check_backpressure().is_ok());
    }

    #[test]
    fn test_backpressure_unlimited_when_zero() {
        let mut mgr = TenantManager::new();
        mgr.set_max_queue_depth(0);
        // Even with huge pending count, should not trigger
        mgr.pending_events.store(1_000_000, Ordering::Relaxed);
        assert!(mgr.check_backpressure().is_ok());
    }

    #[test]
    fn test_queue_pressure_ratio() {
        let mut mgr = TenantManager::new();
        mgr.set_max_queue_depth(100);
        mgr.pending_events.store(50, Ordering::Relaxed);
        let ratio = mgr.queue_pressure_ratio();
        assert!((ratio - 0.5).abs() < f64::EPSILON);
    }

    #[test]
    fn test_queue_pressure_ratio_zero_max() {
        let mgr = TenantManager::new();
        assert_eq!(mgr.queue_pressure_ratio(), 0.0);
    }

    #[test]
    fn test_pending_events_counter() {
        let mgr = TenantManager::new();
        assert_eq!(mgr.pending_event_count(), 0);
        let counter = mgr.pending_events_counter();
        counter.fetch_add(5, Ordering::Relaxed);
        assert_eq!(mgr.pending_event_count(), 5);
    }

    #[tokio::test]
    async fn test_process_event_with_backpressure_ok() {
        let mut mgr = TenantManager::new();
        mgr.set_max_queue_depth(100);
        let id = mgr
            .create_tenant("Test".into(), "key-bp".into(), TenantQuota::default())
            .unwrap();

        let tenant = mgr.get_tenant_mut(&id).unwrap();
        let vpl = "stream A = SensorReading .where(temperature > 100)";
        let pid = tenant
            .deploy_pipeline("BP Pipeline".into(), vpl.into())
            .await
            .unwrap();

        let event = Event::new("SensorReading").with_field("temperature", 150.0);
        let result = mgr.process_event_with_backpressure(&id, &pid, event).await;
        assert!(result.is_ok());
        // Pending count should be back to 0 after processing
        assert_eq!(mgr.pending_event_count(), 0);
    }

    #[tokio::test]
    async fn test_process_event_with_backpressure_rejected() {
        let mut mgr = TenantManager::new();
        mgr.set_max_queue_depth(5);
        // Simulate queue being full
        mgr.pending_events.store(5, Ordering::Relaxed);

        let id = mgr
            .create_tenant("Test".into(), "key-bp2".into(), TenantQuota::default())
            .unwrap();
        let tenant = mgr.get_tenant_mut(&id).unwrap();
        let vpl = "stream A = SensorReading .where(temperature > 100)";
        let pid = tenant
            .deploy_pipeline("BP Pipeline 2".into(), vpl.into())
            .await
            .unwrap();

        let event = Event::new("SensorReading").with_field("temperature", 150.0);
        let result = mgr.process_event_with_backpressure(&id, &pid, event).await;
        assert!(result.is_err());
        match result {
            Err(TenantError::BackpressureExceeded { current, max }) => {
                assert_eq!(current, 5);
                assert_eq!(max, 5);
            }
            _ => panic!("Expected BackpressureExceeded"),
        }
    }

    #[test]
    fn test_backpressure_error_display() {
        let err = TenantError::BackpressureExceeded {
            current: 50000,
            max: 50000,
        };
        let msg = format!("{err}");
        assert!(msg.contains("50000"));
        assert!(msg.contains("exceeds maximum"));
    }

    #[tokio::test]
    async fn test_collect_pipeline_metrics_returns_event_counts() {
        let mut mgr = TenantManager::new();
        let id = mgr
            .create_tenant("Test".into(), "key-1".into(), TenantQuota::default())
            .unwrap();

        // Deploy a simple filter pipeline
        let vpl = "stream A = SensorReading .where(temperature > 100)";
        mgr.deploy_pipeline_on_tenant(&id, "test-pipeline".into(), vpl.into())
            .await
            .unwrap();

        // Verify metrics before any events: pipeline exists with 0 events
        let metrics = mgr.collect_pipeline_metrics().await;
        assert_eq!(metrics.len(), 1, "Should have 1 pipeline");
        assert_eq!(metrics[0].0, "test-pipeline");
        assert_eq!(metrics[0].1, 0, "events_in should be 0 before processing");
        assert_eq!(metrics[0].2, 0, "events_out should be 0 before processing");

        // Process events through the pipeline
        let tenant = mgr.get_tenant_mut(&id).unwrap();
        let pid = tenant.pipelines.keys().next().unwrap().clone();
        for i in 0..10 {
            let event = Event::new("SensorReading")
                .with_field("temperature", (i as f64).mul_add(20.0, 50.0));
            tenant.process_event(&pid, event).await.unwrap();
        }

        // Re-collect metrics — should reflect processed events
        let metrics = mgr.collect_pipeline_metrics().await;
        assert_eq!(metrics.len(), 1);
        assert_eq!(metrics[0].0, "test-pipeline");
        assert!(
            metrics[0].1 > 0,
            "events_in should be > 0 after processing, got {}",
            metrics[0].1
        );
    }

    #[tokio::test]
    async fn test_collect_pipeline_metrics_multiple_tenants() {
        let mut mgr = TenantManager::new();
        let id1 = mgr
            .create_tenant("Tenant1".into(), "key-1".into(), TenantQuota::default())
            .unwrap();
        let id2 = mgr
            .create_tenant("Tenant2".into(), "key-2".into(), TenantQuota::default())
            .unwrap();

        let vpl = "stream A = SensorReading .where(temperature > 100)";
        mgr.deploy_pipeline_on_tenant(&id1, "pipeline-1".into(), vpl.into())
            .await
            .unwrap();
        mgr.deploy_pipeline_on_tenant(&id2, "pipeline-2".into(), vpl.into())
            .await
            .unwrap();

        let metrics = mgr.collect_pipeline_metrics().await;
        assert_eq!(metrics.len(), 2, "Should have 2 pipelines across 2 tenants");
        let names: Vec<&str> = metrics.iter().map(|(n, _, _)| n.as_str()).collect();
        assert!(names.contains(&"pipeline-1"));
        assert!(names.contains(&"pipeline-2"));
    }

    #[tokio::test]
    async fn test_collect_pipeline_metrics_empty_when_no_pipelines() {
        let mut mgr = TenantManager::new();
        let _id = mgr
            .create_tenant("Test".into(), "key-1".into(), TenantQuota::default())
            .unwrap();

        let metrics = mgr.collect_pipeline_metrics().await;
        assert!(
            metrics.is_empty(),
            "Should be empty when no pipelines deployed"
        );
    }

    #[tokio::test]
    async fn test_deploy_pipeline_on_tenant_and_collect_metrics() {
        // Test via TenantManager.deploy_pipeline_on_tenant → collect_pipeline_metrics
        // This mirrors the exact code path used in production (coordinator deploys to worker)
        let mut mgr = TenantManager::new();
        let id = mgr
            .create_tenant(
                "default".into(),
                "api-key-123".into(),
                TenantQuota::enterprise(),
            )
            .unwrap();

        let vpl = r"
            event MarketTick:
                symbol: str
                price: float
            stream Ticks = MarketTick .where(price > 100)
        ";
        let pipeline_id = mgr
            .deploy_pipeline_on_tenant(&id, "financial-cep".into(), vpl.into())
            .await
            .unwrap();

        // Verify the pipeline appears in metrics
        let metrics = mgr.collect_pipeline_metrics().await;
        assert_eq!(metrics.len(), 1);
        assert_eq!(metrics[0].0, "financial-cep");

        // Process events and verify counts update
        for _ in 0..5 {
            let event = Event::new("MarketTick")
                .with_field("symbol", "AAPL")
                .with_field("price", 150.0);
            mgr.process_event_with_backpressure(&id, &pipeline_id, event)
                .await
                .unwrap();
        }

        let metrics = mgr.collect_pipeline_metrics().await;
        assert_eq!(metrics[0].1, 5, "Should have processed 5 events");
    }
}