fraiseql-core 2.15.0

use std::sync::{
    Arc,
    atomic::{AtomicU64, Ordering},
};

use dashmap::DashMap;
use tokio::sync::broadcast;

#[allow(clippy::wildcard_imports)]
// Reason: types::* re-exports the subscription type vocabulary used throughout this module
use super::{SubscriptionError, types::*};
use crate::schema::CompiledSchema;

// =============================================================================
// Subscription Manager
// =============================================================================

/// Maximum number of active subscriptions a single connection may hold.
///
/// Prevents a single authenticated connection from exhausting server memory by
/// calling `subscribe()` in a loop.
const MAX_SUBSCRIPTIONS_PER_CONNECTION: usize = 100;

/// Manages active subscriptions and event routing.
///
/// The `SubscriptionManager` is the central hub for:
/// - Tracking active subscriptions per connection
/// - Receiving events from database listeners
/// - Matching events to subscriptions
/// - Broadcasting to transport adapters
pub struct SubscriptionManager {
    /// Compiled schema for subscription definitions.
    schema: Arc<CompiledSchema>,

    /// Active subscriptions indexed by ID.
    subscriptions: DashMap<SubscriptionId, ActiveSubscription>,

    /// Subscriptions indexed by connection ID (for cleanup on disconnect).
    subscriptions_by_connection: DashMap<String, Vec<SubscriptionId>>,

    /// Broadcast channel for delivering events to transports.
    event_sender: broadcast::Sender<SubscriptionPayload>,

    /// Monotonic sequence counter for event ordering.
    sequence_counter: AtomicU64,

    /// Whether the deployment is multi-tenant (`security.multi_tenant`).
    ///
    /// In multi-tenant mode the tenant gate fails **closed**: an event is only
    /// delivered when both the subscription and the event carry the *same*
    /// tenant — a missing tenant on either side never matches (M-tenant-ws-failopen).
    /// In single-tenant mode the gate stays permissive so existing deployments,
    /// where tenant ids are typically absent, keep working.
    multi_tenant: bool,
}

impl SubscriptionManager {
    /// Create a new subscription manager.
    ///
    /// # Arguments
    ///
    /// * `schema` - Compiled schema containing subscription definitions
    /// * `channel_capacity` - Broadcast channel capacity (default: 1024)
    #[must_use]
    pub fn new(schema: Arc<CompiledSchema>) -> Self {
        Self::with_capacity(schema, 1024)
    }

    /// Create a new subscription manager with custom channel capacity.
    #[must_use]
    pub fn with_capacity(schema: Arc<CompiledSchema>, channel_capacity: usize) -> Self {
        let (event_sender, _) = broadcast::channel(channel_capacity);
        // Derive the tenancy mode from the compiled schema the manager already
        // holds — no separate flag has to be threaded through every call site.
        let multi_tenant = schema.is_multi_tenant();

        Self {
            schema,
            subscriptions: DashMap::new(),
            subscriptions_by_connection: DashMap::new(),
            event_sender,
            sequence_counter: AtomicU64::new(1),
            multi_tenant,
        }
    }

    /// Get a receiver for subscription payloads.
    ///
    /// Transport adapters use this to receive events for delivery.
    #[must_use]
    pub fn receiver(&self) -> broadcast::Receiver<SubscriptionPayload> {
        self.event_sender.subscribe()
    }

    /// Subscribe to a subscription type.
    ///
    /// # Arguments
    ///
    /// * `subscription_name` - Name of the subscription type
    /// * `user_context` - User authentication/authorization context
    /// * `variables` - Runtime variables from client
    /// * `connection_id` - Client connection identifier
    ///
    /// # Errors
    ///
    /// Returns error if subscription not found or user not authorized.
    pub fn subscribe(
        &self,
        subscription_name: &str,
        user_context: serde_json::Value,
        variables: serde_json::Value,
        connection_id: &str,
    ) -> Result<SubscriptionId, SubscriptionError> {
        self.subscribe_with_rls(
            subscription_name,
            user_context,
            variables,
            connection_id,
            Vec::new(),
        )
    }

    /// Subscribe with pre-evaluated RLS conditions for event-level filtering.
    ///
    /// The caller should evaluate the RLS policy at subscribe time and pass
    /// the resulting conditions (via [`extract_rls_conditions`](super::extract_rls_conditions)).
    /// During event delivery, each condition is checked against the event data:
    /// the event is only delivered if **every** condition matches.
    ///
    /// # Errors
    ///
    /// Returns error if subscription not found or connection limit exceeded.
    pub fn subscribe_with_rls(
        &self,
        subscription_name: &str,
        user_context: serde_json::Value,
        variables: serde_json::Value,
        connection_id: &str,
        rls_conditions: Vec<(String, serde_json::Value)>,
    ) -> Result<SubscriptionId, SubscriptionError> {
        self.register(
            subscription_name,
            user_context,
            variables,
            connection_id,
            rls_conditions,
            None,
        )
    }

    /// Subscribe a subscriber whose events are served through `plan` (ruling AA 4): the
    /// executor planned it at subscribe time from the client's selection and principal
    /// ([`Executor::plan_subscription`](crate::runtime::Executor::plan_subscription)). An event
    /// the plan suppresses is not delivered, and nothing tells the subscriber it existed.
    ///
    /// # Errors
    ///
    /// As [`subscribe_with_rls`](Self::subscribe_with_rls).
    pub fn subscribe_planned(
        &self,
        plan: Arc<crate::runtime::SubscriptionPlan>,
        user_context: serde_json::Value,
        variables: serde_json::Value,
        connection_id: &str,
        rls_conditions: Vec<(String, serde_json::Value)>,
    ) -> Result<SubscriptionId, SubscriptionError> {
        let name = plan.subscription_name().to_string();
        self.register(&name, user_context, variables, connection_id, rls_conditions, Some(plan))
    }

    fn register(
        &self,
        subscription_name: &str,
        user_context: serde_json::Value,
        variables: serde_json::Value,
        connection_id: &str,
        rls_conditions: Vec<(String, serde_json::Value)>,
        plan: Option<Arc<crate::runtime::SubscriptionPlan>>,
    ) -> Result<SubscriptionId, SubscriptionError> {
        // Find subscription definition
        let mut definition = self
            .schema
            .find_subscription(subscription_name)
            .ok_or_else(|| SubscriptionError::SubscriptionNotFound(subscription_name.to_string()))?
            .clone();

        // Expand filter_fields into argument_paths on the filter.
        // Each filter_field name becomes an argument_path entry mapping
        // the field name to a JSON pointer path (e.g., "user_id" → "/user_id").
        if !definition.filter_fields.is_empty() {
            let filter =
                definition.filter.get_or_insert_with(|| crate::schema::SubscriptionFilter {
                    argument_paths: std::collections::HashMap::new(),
                    static_filters: Vec::new(),
                });
            for field in &definition.filter_fields {
                filter
                    .argument_paths
                    .entry(field.clone())
                    .or_insert_with(|| format!("/{field}"));
            }
        }

        // Every filter key must name a declared argument (#1262). Checked here, after the
        // expansion above, so a `filter_fields` entry is covered by the same rule as an
        // authored condition. `fraiseql compile` refuses this shape, so reaching it means
        // the compiled schema was edited by hand.
        Self::check_filter_arguments(&definition)?;

        // Ruling AC 6: a type with a read gate is delivered only through a plan, which the
        // executor builds from the subscriber's selection and principal. Without one, the
        // manager would serve every field of every after-image.
        if plan.is_none() {
            if let Some(gate) = self.read_gate(&definition.return_type) {
                return Err(SubscriptionError::Forbidden(format!(
                    "Subscription '{subscription_name}' delivers '{}', which {gate}: it is \
                     delivered only through a plan built from the subscriber's selection \
                     and principal (`Executor::plan_subscription`)",
                    definition.return_type
                )));
            }
        }

        // Create active subscription with RLS conditions
        let mut active = ActiveSubscription::new(
            subscription_name,
            definition,
            user_context,
            variables,
            connection_id,
        )
        .with_rls_conditions(rls_conditions);
        if let Some(plan) = plan {
            active = active.with_plan(plan);
        }

        let id = active.id;

        // Enforce per-connection subscription cap before inserting.
        {
            let mut conn_subs =
                self.subscriptions_by_connection.entry(connection_id.to_string()).or_default();
            if conn_subs.len() >= MAX_SUBSCRIPTIONS_PER_CONNECTION {
                return Err(SubscriptionError::Internal(format!(
                    "Connection '{connection_id}' has reached the maximum of \
                     {MAX_SUBSCRIPTIONS_PER_CONNECTION} concurrent subscriptions"
                )));
            }
            conn_subs.push(id);
        }

        // Store subscription
        self.subscriptions.insert(id, active);

        tracing::info!(
            subscription_id = %id,
            subscription_name = subscription_name,
            connection_id = connection_id,
            "Subscription created"
        );

        Ok(id)
    }

    /// Refuse a subscription whose filter names an argument it does not declare (#1262).
    ///
    /// Call sites pass a definition whose `filter_fields` have already been expanded into
    /// `argument_paths`, so this sees the same key set the delivery loop will look up.
    ///
    /// # Errors
    ///
    /// Returns [`SubscriptionError::UnresolvableFilter`] naming every unresolvable
    /// reference and the arguments the subscription does declare.
    /// The first read gate a document of `type_name` can meet, at any level: a type's
    /// `requires_role`, a `requires_scope` field, an `authorize` field.
    fn read_gate(&self, type_name: &str) -> Option<String> {
        let mut seen = std::collections::HashSet::new();
        let mut pending = vec![type_name.to_string()];
        while let Some(current) = pending.pop() {
            if !seen.insert(current.clone()) {
                continue;
            }
            let Some(type_def) = self.schema.find_type(&current) else {
                continue;
            };
            if type_def.requires_role.is_some() {
                return Some(format!("reaches '{current}', whose read requires a role"));
            }
            for field in &type_def.fields {
                if field.requires_scope.is_some() || field.authorize {
                    return Some(format!("reaches the gated field '{current}.{}'", field.name));
                }
                if let Some(child) =
                    field.field_type.inner_type().unwrap_or(&field.field_type).type_name()
                {
                    pending.push(child.to_string());
                }
            }
        }
        None
    }

    fn check_filter_arguments(
        definition: &crate::schema::SubscriptionDefinition,
    ) -> Result<(), SubscriptionError> {
        definition
            .filter_violation()
            .map_or(Ok(()), |violation| Err(SubscriptionError::UnresolvableFilter(violation)))
    }

    /// Unsubscribe from a subscription.
    ///
    /// # Errors
    ///
    /// Returns error if subscription not found.
    pub fn unsubscribe(&self, id: SubscriptionId) -> Result<(), SubscriptionError> {
        let removed = self
            .subscriptions
            .remove(&id)
            .ok_or_else(|| SubscriptionError::NotActive(id.to_string()))?;

        // Remove from connection index
        if let Some(mut subs) = self.subscriptions_by_connection.get_mut(&removed.1.connection_id) {
            subs.retain(|s| *s != id);
        }

        tracing::info!(
            subscription_id = %id,
            subscription_name = removed.1.subscription_name,
            "Subscription removed"
        );

        Ok(())
    }

    /// Unsubscribe all subscriptions for a connection.
    ///
    /// Called when a client disconnects.
    ///
    /// # Concurrency note
    ///
    /// A concurrent `subscribe` call that runs between the `DashMap` entry removal and the
    /// per-subscription cleanup loop would create a new connection entry that is not cleaned
    /// up by this call. A second-pass removal after the first loop closes this window for
    /// all but the most extreme concurrent races. Any subscription that slips through is
    /// benign: it will receive events until the broadcast receiver is dropped (which happens
    /// on disconnect), and will be removed on the next disconnect or subscription-not-found
    /// event for that ID.
    pub fn unsubscribe_connection(&self, connection_id: &str) {
        // First pass: remove the connection index atomically and clean up known subscriptions.
        let first_pass_count = if let Some((_, subscription_ids)) =
            self.subscriptions_by_connection.remove(connection_id)
        {
            let count = subscription_ids.len();
            for id in subscription_ids {
                self.subscriptions.remove(&id);
            }
            count
        } else {
            0
        };

        // Second pass: clean up any subscriptions added by a concurrent `subscribe` call that
        // ran between the `remove()` above and the loop.  A concurrent `subscribe` that saw
        // the connection entry absent would have inserted a *new* entry; removing it here
        // closes the TOCTOU window to a negligible two-CAS race.
        let second_pass_count = if let Some((_, subscription_ids)) =
            self.subscriptions_by_connection.remove(connection_id)
        {
            let count = subscription_ids.len();
            for id in subscription_ids {
                self.subscriptions.remove(&id);
                tracing::warn!(
                    subscription_id = %id,
                    connection_id = connection_id,
                    "Cleaned up subscription added concurrently during disconnect"
                );
            }
            count
        } else {
            0
        };

        tracing::info!(
            connection_id = connection_id,
            subscriptions_removed = first_pass_count + second_pass_count,
            "All subscriptions removed for connection"
        );
    }

    /// Get an active subscription by ID.
    #[must_use]
    pub fn get_subscription(&self, id: SubscriptionId) -> Option<ActiveSubscription> {
        self.subscriptions.get(&id).map(|r| r.clone())
    }

    /// Replace the plan every later event of subscription `id` is served through — a policy
    /// reload re-plans a live subscription (ruling AC 7).
    ///
    /// # Errors
    ///
    /// [`SubscriptionError::NotActive`] when the subscription is gone.
    pub fn replace_plan(
        &self,
        id: SubscriptionId,
        plan: Arc<crate::runtime::SubscriptionPlan>,
    ) -> Result<(), SubscriptionError> {
        let mut subscription = self
            .subscriptions
            .get_mut(&id)
            .ok_or_else(|| SubscriptionError::NotActive(id.to_string()))?;
        subscription.plan = Some(plan);
        Ok(())
    }

    /// Replace the server-owned RLS conditions of a live subscription (#611).
    ///
    /// Used when a schema hot-reload changes the row-visibility policy behind an
    /// already-connected subscription: the transport re-derives the conditions
    /// from the new policy and swaps them in, so the tightened boundary applies
    /// to the next delivered event without waiting for a reconnect. Delivery
    /// matching reads the conditions per event, so the swap takes effect
    /// immediately and atomically.
    ///
    /// # Errors
    ///
    /// Returns [`SubscriptionError::NotActive`] when the subscription no longer
    /// exists (e.g. it raced a disconnect) — callers treat that as already gone.
    pub fn update_rls_conditions(
        &self,
        id: SubscriptionId,
        rls_conditions: Vec<(String, serde_json::Value)>,
    ) -> Result<(), SubscriptionError> {
        let mut subscription = self
            .subscriptions
            .get_mut(&id)
            .ok_or_else(|| SubscriptionError::NotActive(id.to_string()))?;
        subscription.rls_conditions = rls_conditions;
        Ok(())
    }

    /// Get all active subscriptions for a connection.
    #[must_use]
    pub fn get_connection_subscriptions(&self, connection_id: &str) -> Vec<ActiveSubscription> {
        self.subscriptions_by_connection
            .get(connection_id)
            .map(|ids| {
                ids.iter()
                    .filter_map(|id| self.subscriptions.get(id).map(|r| r.clone()))
                    .collect()
            })
            .unwrap_or_default()
    }

    /// Get total number of active subscriptions.
    #[must_use]
    pub fn subscription_count(&self) -> usize {
        self.subscriptions.len()
    }

    /// Get number of active connections with subscriptions.
    #[must_use]
    pub fn connection_count(&self) -> usize {
        self.subscriptions_by_connection.len()
    }

    /// Publish an event to matching subscriptions.
    ///
    /// This is called by the database listener when an event is received.
    /// The event is matched against all active subscriptions and delivered
    /// to matching ones.
    ///
    /// # Arguments
    ///
    /// * `event` - The database event to publish
    ///
    /// # Returns
    ///
    /// Number of subscriptions that matched the event.
    pub fn publish_event(&self, mut event: SubscriptionEvent) -> usize {
        // Assign sequence number
        event.sequence_number = self.sequence_counter.fetch_add(1, Ordering::SeqCst);

        let mut matched = 0;

        // Find matching subscriptions
        for subscription in &self.subscriptions {
            if self.matches_subscription(&event, &subscription) {
                // A planned subscription is served through its plan; an event the plan
                // suppresses is not delivered, and counts as no match.
                let data = match &subscription.plan {
                    Some(plan) => match plan.deliver(&event.data) {
                        Some(served) => served,
                        None => continue,
                    },
                    None => self.project_event_data(&event, &subscription),
                };
                matched += 1;

                // Every seam reads this payload — `/ws` its `data`, the Kafka mirror and the
                // webhook adapter its `event` (ruling AC 7). For a planned subscription the
                // event carries the served document and no before-image, which no plan
                // covers: nothing downstream of the plan can see the raw after-image.
                let mut delivered = event.clone();
                if subscription.plan.is_some() {
                    delivered.data = data.clone();
                    delivered.old_data = None;
                }
                // Ruling AA 5: the subscriber sees its own delivery position, gap-free —
                // never the manager's global counter or the Change-Spine `seq`, which count
                // every change, those withheld from this subscriber included.
                delivered.sequence_number =
                    subscription.delivered.fetch_add(1, Ordering::SeqCst) + 1;
                // Ruling AI: the envelope names no principal but the subscriber and carries
                // nothing rows it may not read shape (`seq` included).
                let own_user = subscription.plan.as_ref().and_then(|plan| plan.subscriber_uuid());
                delivered.change_spine = delivered.change_spine.as_ref().and_then(|envelope| {
                    envelope.for_subscriber(own_user, subscription.tenant_id.as_deref())
                });
                let payload = SubscriptionPayload {
                    subscription_id: subscription.id,
                    subscription_name: subscription.subscription_name.clone(),
                    event: delivered,
                    data,
                };

                // Send to broadcast channel (may fail if no receivers, that's ok)
                let _ = self.event_sender.send(payload);
            }
        }

        if matched > 0 {
            tracing::debug!(
                event_id = event.event_id,
                entity_type = event.entity_type,
                operation = %event.operation,
                matched = matched,
                "Event matched subscriptions"
            );
        }

        matched
    }

    /// Check if an event matches a subscription's filters and RLS conditions.
    #[allow(clippy::cognitive_complexity)] // Reason: multi-criteria subscription matching (entity type, operation, filters, RLS conditions)
    fn matches_subscription(
        &self,
        event: &SubscriptionEvent,
        subscription: &ActiveSubscription,
    ) -> bool {
        // Tenant isolation (M-tenant-ws-failopen).
        //
        // Multi-tenant deployments fail **closed**: an event is delivered only
        // when the subscription and the event carry the *same* tenant. A missing
        // tenant on either side never matches — in particular a tenant-less
        // subscriber must not receive any tenant's events (the fail-open hole),
        // and a tenant-scoped subscriber must not receive untagged events.
        //
        // Single-tenant deployments stay permissive: tenant ids are typically
        // absent, so the gate only filters out events that carry a *different*
        // explicit tenant than a tenant-scoped subscription.
        let tenant_ok = match (subscription.tenant_id.as_deref(), event.tenant_id.as_deref()) {
            (Some(sub_tenant), Some(event_tenant)) => sub_tenant == event_tenant,
            // One or both sides lack a tenant: only acceptable in single-tenant mode.
            _ => !self.multi_tenant,
        };
        if !tenant_ok {
            tracing::trace!(
                subscription_id = %subscription.id,
                sub_tenant = ?subscription.tenant_id,
                event_tenant = ?event.tenant_id,
                multi_tenant = self.multi_tenant,
                "Tenant gate — event filtered"
            );
            return false;
        }

        // Check entity type matches (subscription return_type maps to entity)
        if subscription.definition.return_type != event.entity_type {
            return false;
        }

        // Check operation matches topic (if specified)
        if let Some(ref topic) = subscription.definition.topic {
            let expected_op = match topic.to_lowercase().as_str() {
                t if t.contains("created") || t.contains("insert") => {
                    Some(SubscriptionOperation::Create)
                },
                t if t.contains("updated") || t.contains("update") => {
                    Some(SubscriptionOperation::Update)
                },
                t if t.contains("deleted") || t.contains("delete") => {
                    Some(SubscriptionOperation::Delete)
                },
                _ => None,
            };

            if let Some(expected) = expected_op {
                if event.operation != expected {
                    return false;
                }
            }
        }

        // Check row-level security conditions (evaluated at subscribe time).
        // Every condition must match (AND semantics) — RLS always wins.
        for (field, expected_value) in &subscription.rls_conditions {
            let actual = get_json_pointer_value(&event.data, field);
            if actual != Some(expected_value) {
                tracing::trace!(
                    subscription_id = %subscription.id,
                    field = field,
                    expected = ?expected_value,
                    actual = ?actual,
                    "RLS condition mismatch — event filtered"
                );
                return false;
            }
        }

        // Evaluate compiled WHERE filters against event.data and subscription variables
        if let Some(ref filter) = subscription.definition.filter {
            // Check argument-based filters (variable values must match event data)
            for (arg_name, path) in &filter.argument_paths {
                // Get the variable value provided by the client
                if let Some(expected_value) = subscription.variables.get(arg_name) {
                    // Get the actual value from event data using JSON pointer
                    let actual_value = get_json_pointer_value(&event.data, path);

                    // Compare values
                    if actual_value != Some(expected_value) {
                        tracing::trace!(
                            subscription_id = %subscription.id,
                            arg_name = arg_name,
                            expected = ?expected_value,
                            actual = ?actual_value,
                            "Filter mismatch on argument"
                        );
                        return false;
                    }
                }
            }

            // Check static filter conditions
            for condition in &filter.static_filters {
                let actual_value = get_json_pointer_value(&event.data, &condition.path);

                if !evaluate_filter_condition(actual_value, condition.operator, &condition.value) {
                    tracing::trace!(
                        subscription_id = %subscription.id,
                        path = condition.path,
                        operator = ?condition.operator,
                        expected = ?condition.value,
                        actual = ?actual_value,
                        "Filter mismatch on static condition"
                    );
                    return false;
                }
            }
        }

        true
    }

    /// Project event data to subscription's field selection.
    fn project_event_data(
        &self,
        event: &SubscriptionEvent,
        subscription: &ActiveSubscription,
    ) -> serde_json::Value {
        // If no fields specified, return full event data
        if subscription.definition.fields.is_empty() {
            return event.data.clone();
        }

        // Project only requested fields
        let mut projected = serde_json::Map::new();

        for field in &subscription.definition.fields {
            // Support both simple field names and JSON pointer paths
            let value = if field.starts_with('/') {
                get_json_pointer_value(&event.data, field).cloned()
            } else {
                event.data.get(field).cloned()
            };

            if let Some(v) = value {
                // Use the field name (without leading slash) as the key
                let key = field.trim_start_matches('/').to_string();
                projected.insert(key, v);
            }
        }

        serde_json::Value::Object(projected)
    }
}

/// Retrieve a value from JSON data using a JSON pointer path.
///
/// # Lifetime Parameter
///
/// The lifetime `'a` is tied to the input `data` reference. The returned reference
/// is guaranteed to live as long as the input data reference, enabling zero-copy
/// access to nested JSON values without allocation.
///
/// # Arguments
///
/// * `data` - The JSON data object to query
/// * `path` - The path to the value, either in JSON pointer format (/a/b/c) or dot notation (a.b.c)
///
/// # Returns
///
/// A reference to the JSON value if found, or `None` if the path doesn't exist.
/// The returned reference has the same lifetime as the input data.
///
/// # Examples
///
/// ```rust
/// # use serde_json::json;
/// # fn get_json_pointer_value<'a>(data: &'a serde_json::Value, path: &str) -> Option<&'a serde_json::Value> {
/// #     let normalized = if path.starts_with('/') { path.to_string() } else { format!("/{}", path.replace('.', "/")) };
/// #     data.pointer(&normalized)
/// # }
/// let data = json!({"user": {"id": 123, "name": "Alice"}});
/// let id = get_json_pointer_value(&data, "user/id");  // Some(&123)
/// let alt = get_json_pointer_value(&data, "user.id"); // Some(&123)
/// let missing = get_json_pointer_value(&data, "admin/id"); // None
/// ```
pub fn get_json_pointer_value<'a>(
    data: &'a serde_json::Value,
    path: &str,
) -> Option<&'a serde_json::Value> {
    // Normalize path to JSON pointer format
    let normalized = if path.starts_with('/') {
        path.to_string()
    } else {
        format!("/{}", path.replace('.', "/"))
    };

    data.pointer(&normalized)
}

/// Evaluate a filter condition against an actual value.
pub fn evaluate_filter_condition(
    actual: Option<&serde_json::Value>,
    operator: crate::schema::FilterOperator,
    expected: &serde_json::Value,
) -> bool {
    use crate::schema::FilterOperator;

    match actual {
        None => {
            // Null/missing values only match specific conditions
            matches!(operator, FilterOperator::Eq) && expected.is_null()
        },
        Some(actual_value) => match operator {
            FilterOperator::Eq => actual_value == expected,
            FilterOperator::Ne => actual_value != expected,
            FilterOperator::Gt => {
                compare_values(actual_value, expected) == Some(std::cmp::Ordering::Greater)
            },
            FilterOperator::Gte => {
                matches!(
                    compare_values(actual_value, expected),
                    Some(std::cmp::Ordering::Greater | std::cmp::Ordering::Equal)
                )
            },
            FilterOperator::Lt => {
                compare_values(actual_value, expected) == Some(std::cmp::Ordering::Less)
            },
            FilterOperator::Lte => {
                matches!(
                    compare_values(actual_value, expected),
                    Some(std::cmp::Ordering::Less | std::cmp::Ordering::Equal)
                )
            },
            FilterOperator::Contains => {
                match (actual_value, expected) {
                    // Array contains value
                    (serde_json::Value::Array(arr), val) => arr.contains(val),
                    // String contains substring
                    (serde_json::Value::String(s), serde_json::Value::String(sub)) => {
                        s.contains(sub.as_str())
                    },
                    _ => false,
                }
            },
            FilterOperator::StartsWith => match (actual_value, expected) {
                (serde_json::Value::String(s), serde_json::Value::String(prefix)) => {
                    s.starts_with(prefix.as_str())
                },
                _ => false,
            },
            FilterOperator::EndsWith => match (actual_value, expected) {
                (serde_json::Value::String(s), serde_json::Value::String(suffix)) => {
                    s.ends_with(suffix.as_str())
                },
                _ => false,
            },
        },
    }
}

/// Compare two JSON values for ordering (numeric and string comparisons).
fn compare_values(a: &serde_json::Value, b: &serde_json::Value) -> Option<std::cmp::Ordering> {
    match (a, b) {
        // Numeric comparisons
        (serde_json::Value::Number(a), serde_json::Value::Number(b)) => {
            let a_f64 = a.as_f64()?;
            let b_f64 = b.as_f64()?;
            a_f64.partial_cmp(&b_f64)
        },
        // String comparisons
        (serde_json::Value::String(a), serde_json::Value::String(b)) => Some(a.cmp(b)),
        // Bool comparisons (false < true)
        (serde_json::Value::Bool(a), serde_json::Value::Bool(b)) => Some(a.cmp(b)),
        // Null comparisons
        (serde_json::Value::Null, serde_json::Value::Null) => Some(std::cmp::Ordering::Equal),
        // Incompatible types
        _ => None,
    }
}

impl std::fmt::Debug for SubscriptionManager {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("SubscriptionManager")
            .field("subscription_count", &self.subscriptions.len())
            .field("connection_count", &self.subscriptions_by_connection.len())
            .finish_non_exhaustive()
    }
}