#![allow(clippy::unwrap_used)] // Reason: test code, panics are acceptable
use std::sync::Arc;

use uuid::Uuid;

use super::*;
use crate::schema::{CompiledSchema, SecurityConfig, SubscriptionDefinition};

fn create_test_schema() -> CompiledSchema {
    CompiledSchema {
        subscriptions: vec![
            SubscriptionDefinition::new("OrderCreated", "Order").with_topic("order_created"),
            SubscriptionDefinition::new("OrderUpdated", "Order").with_topic("order_updated"),
            SubscriptionDefinition::new("UserDeleted", "User").with_topic("user_deleted"),
        ],
        ..Default::default()
    }
}

/// Same subscriptions as [`create_test_schema`] but flagged multi-tenant
/// (`security.multi_tenant = true`), so the manager's tenant gate fails closed.
fn create_multi_tenant_schema() -> CompiledSchema {
    let mut schema = create_test_schema();
    schema.security = Some(SecurityConfig {
        multi_tenant: true,
        ..Default::default()
    });
    schema
}

#[test]
fn test_subscription_id() {
    let id1 = SubscriptionId::new();
    let id2 = SubscriptionId::new();
    assert_ne!(id1, id2);

    let uuid = Uuid::new_v4();
    let id3 = SubscriptionId::from_uuid(uuid);
    assert_eq!(id3.0, uuid);
}

#[test]
fn test_subscription_event_creation() {
    let event = SubscriptionEvent::new(
        "Order",
        "ord_123",
        SubscriptionOperation::Create,
        serde_json::json!({"id": "ord_123", "amount": 99.99}),
    );

    assert!(event.event_id.starts_with("evt_"));
    assert_eq!(event.entity_type, "Order");
    assert_eq!(event.entity_id, "ord_123");
    assert_eq!(event.operation, SubscriptionOperation::Create);
}

#[test]
fn test_subscription_manager_subscribe() {
    let schema = Arc::new(create_test_schema());
    let manager = SubscriptionManager::new(schema);

    let id = manager
        .subscribe(
            "OrderCreated",
            serde_json::json!({"user_id": "usr_123"}),
            serde_json::json!({}),
            "conn_1",
        )
        .unwrap();

    assert_eq!(manager.subscription_count(), 1);
    assert_eq!(manager.connection_count(), 1);

    let sub = manager.get_subscription(id).unwrap();
    assert_eq!(sub.subscription_name, "OrderCreated");
    assert_eq!(sub.connection_id, "conn_1");
}

#[test]
fn test_subscription_manager_subscribe_not_found() {
    let schema = Arc::new(create_test_schema());
    let manager = SubscriptionManager::new(schema);

    let result =
        manager.subscribe("NonExistent", serde_json::json!({}), serde_json::json!({}), "conn_1");

    assert!(matches!(result, Err(SubscriptionError::SubscriptionNotFound(_))));
}

#[test]
fn test_subscription_manager_unsubscribe() {
    let schema = Arc::new(create_test_schema());
    let manager = SubscriptionManager::new(schema);

    let id = manager
        .subscribe("OrderCreated", serde_json::json!({}), serde_json::json!({}), "conn_1")
        .unwrap();

    assert_eq!(manager.subscription_count(), 1);

    manager.unsubscribe(id).unwrap();

    assert_eq!(manager.subscription_count(), 0);
}

#[test]
fn test_subscription_manager_unsubscribe_connection() {
    let schema = Arc::new(create_test_schema());
    let manager = SubscriptionManager::new(schema);

    // Create multiple subscriptions for same connection
    manager
        .subscribe("OrderCreated", serde_json::json!({}), serde_json::json!({}), "conn_1")
        .unwrap();

    manager
        .subscribe("OrderUpdated", serde_json::json!({}), serde_json::json!({}), "conn_1")
        .unwrap();

    assert_eq!(manager.subscription_count(), 2);

    manager.unsubscribe_connection("conn_1");

    assert_eq!(manager.subscription_count(), 0);
    assert_eq!(manager.connection_count(), 0);
}

#[test]
fn test_subscription_event_matching() {
    let schema = Arc::new(create_test_schema());
    let manager = SubscriptionManager::new(schema);

    // Subscribe to OrderCreated
    manager
        .subscribe("OrderCreated", serde_json::json!({}), serde_json::json!({}), "conn_1")
        .unwrap();

    // Create event should match
    let create_event = SubscriptionEvent::new(
        "Order",
        "ord_123",
        SubscriptionOperation::Create,
        serde_json::json!({"id": "ord_123"}),
    );

    let delivered = manager.publish_event(create_event);
    assert_eq!(delivered, 1);

    // Update event should not match (wrong operation)
    let update_event = SubscriptionEvent::new(
        "Order",
        "ord_123",
        SubscriptionOperation::Update,
        serde_json::json!({"id": "ord_123"}),
    );

    let delivered = manager.publish_event(update_event);
    assert_eq!(delivered, 0);
}

#[test]
fn test_subscription_event_wrong_entity() {
    let schema = Arc::new(create_test_schema());
    let manager = SubscriptionManager::new(schema);

    // Subscribe to OrderCreated
    manager
        .subscribe("OrderCreated", serde_json::json!({}), serde_json::json!({}), "conn_1")
        .unwrap();

    // User event should not match (wrong entity)
    let user_event = SubscriptionEvent::new(
        "User",
        "usr_123",
        SubscriptionOperation::Create,
        serde_json::json!({"id": "usr_123"}),
    );

    let delivered = manager.publish_event(user_event);
    assert_eq!(delivered, 0);
}

#[test]
fn test_subscription_sequence_numbers() {
    let schema = Arc::new(create_test_schema());
    let manager = SubscriptionManager::new(schema);

    manager
        .subscribe("OrderCreated", serde_json::json!({}), serde_json::json!({}), "conn_1")
        .unwrap();

    let mut receiver = manager.receiver();

    // Publish multiple events
    for i in 1..=3 {
        let event = SubscriptionEvent::new(
            "Order",
            format!("ord_{i}"),
            SubscriptionOperation::Create,
            serde_json::json!({"id": format!("ord_{}", i)}),
        );
        manager.publish_event(event);
    }

    // Check sequence numbers are monotonic
    let mut last_seq = 0;
    for _ in 0..3 {
        if let Ok(payload) = receiver.try_recv() {
            assert!(payload.event.sequence_number > last_seq);
            last_seq = payload.event.sequence_number;
        }
    }
}

// =========================================================================

// =========================================================================
// Transport Adapter Tests
// =========================================================================

#[test]
fn test_webhook_config_builder() {
    let config = WebhookConfig::new("https://api.example.com/webhooks")
        .with_secret("my-secret")
        .with_timeout(10_000)
        .with_max_retries(5)
        .with_retry_delay(500)
        .with_header("X-Custom-Header", "custom-value");

    assert_eq!(config.url, "https://api.example.com/webhooks");
    assert_eq!(config.secret, Some("my-secret".to_string()));
    assert_eq!(config.timeout_ms, 10_000);
    assert_eq!(config.max_retries, 5);
    assert_eq!(config.retry_delay_ms, 500);
    assert_eq!(config.headers.get("X-Custom-Header"), Some(&"custom-value".to_string()));
}

#[test]
fn test_webhook_config_defaults() {
    let config = WebhookConfig::new("https://api.example.com/webhooks");

    assert_eq!(config.url, "https://api.example.com/webhooks");
    assert!(config.secret.is_none());
    assert_eq!(config.timeout_ms, 30_000);
    assert_eq!(config.max_retries, 3);
    assert_eq!(config.retry_delay_ms, 1000);
    assert!(config.headers.is_empty());
}

#[test]
fn test_webhook_payload_from_event() {
    let event = SubscriptionEvent {
        event_id:        "evt_123".to_string(),
        entity_type:     "Order".to_string(),
        entity_id:       "ord_456".to_string(),
        operation:       SubscriptionOperation::Create,
        data:            serde_json::json!({"id": "ord_456", "total": 99.99}),
        old_data:        None,
        timestamp:       chrono::Utc::now(),
        sequence_number: 42,
        tenant_id:       None,
        change_spine:    None,
    };

    let payload = WebhookPayload::from_event(&event, "order_created");

    assert_eq!(payload.event_id, "evt_123");
    assert_eq!(payload.subscription_name, "order_created");
    assert_eq!(payload.entity_type, "Order");
    assert_eq!(payload.entity_id, "ord_456");
    assert_eq!(payload.operation, "Create");
    assert_eq!(payload.data["total"], 99.99);
    assert!(payload.old_data.is_none());
    assert_eq!(payload.sequence_number, 42);
}

#[test]
fn test_webhook_adapter_debug() {
    let config = WebhookConfig::new("https://api.example.com/webhooks").with_secret("secret-key");
    let adapter = WebhookAdapter::new(config).unwrap();

    let debug = format!("{:?}", adapter);
    assert!(debug.contains("WebhookAdapter"));
    assert!(debug.contains("https://api.example.com/webhooks"));
    assert!(debug.contains("has_secret: true"));
}

#[test]
fn test_webhook_adapter_name() {
    let config = WebhookConfig::new("https://api.example.com/webhooks");
    let adapter = WebhookAdapter::new(config).unwrap();

    assert_eq!(adapter.name(), "webhook");
}

#[test]
fn test_kafka_config_builder() {
    let config = KafkaConfig::new("localhost:9092", "events")
        .with_client_id("test-client")
        .with_acks("all")
        .with_timeout(5_000)
        .with_compression("gzip");

    assert_eq!(config.brokers, "localhost:9092");
    assert_eq!(config.default_topic, "events");
    assert_eq!(config.client_id, "test-client");
    assert_eq!(config.acks, "all");
    assert_eq!(config.timeout_ms, 5_000);
    assert_eq!(config.compression, Some("gzip".to_string()));
}

#[test]
fn test_kafka_config_defaults() {
    let config = KafkaConfig::new("localhost:9092", "events");

    assert_eq!(config.brokers, "localhost:9092");
    assert_eq!(config.default_topic, "events");
    assert_eq!(config.client_id, "fraiseql");
    assert_eq!(config.acks, "all"); // Default: wait for all replicas
    assert_eq!(config.timeout_ms, 30_000); // 30 seconds default
    assert!(config.compression.is_none());
}

#[test]
fn test_kafka_message_from_event() {
    let event = SubscriptionEvent {
        event_id:        "evt_789".to_string(),
        entity_type:     "User".to_string(),
        entity_id:       "usr_123".to_string(),
        operation:       SubscriptionOperation::Update,
        data:            serde_json::json!({"id": "usr_123", "name": "John"}),
        old_data:        Some(serde_json::json!({"id": "usr_123", "name": "Jane"})),
        timestamp:       chrono::Utc::now(),
        sequence_number: 100,
        tenant_id:       None,
        change_spine:    None,
    };

    let message = KafkaMessage::from_event(&event, "user_updated");

    assert_eq!(message.event_id, "evt_789");
    assert_eq!(message.subscription_name, "user_updated");
    assert_eq!(message.entity_type, "User");
    assert_eq!(message.entity_id, "usr_123");
    assert_eq!(message.operation, "Update");
    assert_eq!(message.data["name"], "John");
    assert_eq!(message.old_data.as_ref().unwrap()["name"], "Jane");
    assert_eq!(message.sequence_number, 100);
}

#[test]
fn test_kafka_message_key() {
    let event = SubscriptionEvent {
        event_id:        "evt_1".to_string(),
        entity_type:     "Order".to_string(),
        entity_id:       "ord_partition_key".to_string(),
        operation:       SubscriptionOperation::Create,
        data:            serde_json::json!({}),
        old_data:        None,
        timestamp:       chrono::Utc::now(),
        sequence_number: 1,
        tenant_id:       None,
        change_spine:    None,
    };

    let message = KafkaMessage::from_event(&event, "test_sub");

    // Key should be entity_id for consistent partitioning
    assert_eq!(message.key(), "ord_partition_key");
}

#[test]
fn test_kafka_adapter_name() {
    let config = KafkaConfig::new("localhost:9092", "events");
    let adapter = KafkaAdapter::new(config).unwrap();

    assert_eq!(adapter.name(), "kafka");
}

#[test]
fn test_transport_manager_new() {
    let manager = TransportManager::new();
    assert!(manager.is_empty());
    assert_eq!(manager.adapter_count(), 0);
}

#[test]
fn test_transport_manager_add_adapter() {
    let mut manager = TransportManager::new();

    let webhook =
        WebhookAdapter::new(WebhookConfig::new("https://api.example.com/webhooks")).unwrap();
    manager.add_adapter(Box::new(webhook));

    assert!(!manager.is_empty());
    assert_eq!(manager.adapter_count(), 1);
}

#[test]
fn test_transport_manager_debug() {
    let mut manager = TransportManager::new();
    let webhook =
        WebhookAdapter::new(WebhookConfig::new("https://api.example.com/webhooks")).unwrap();
    manager.add_adapter(Box::new(webhook));

    let debug = format!("{:?}", manager);
    assert!(debug.contains("TransportManager"));
    assert!(debug.contains("adapter_count: 1"));
}

#[test]
fn test_delivery_result_all_succeeded() {
    let result = DeliveryResult {
        successful: 3,
        failed:     0,
        errors:     vec![],
    };

    assert!(result.all_succeeded());
    assert!(result.any_succeeded());
}

#[test]
fn test_delivery_result_partial_failure() {
    let result = DeliveryResult {
        successful: 2,
        failed:     1,
        errors:     vec![("webhook".to_string(), "Connection refused".to_string())],
    };

    assert!(!result.all_succeeded());
    assert!(result.any_succeeded());
}

#[test]
fn test_delivery_result_all_failed() {
    let result = DeliveryResult {
        successful: 0,
        failed:     2,
        errors:     vec![
            ("webhook".to_string(), "Connection refused".to_string()),
            ("kafka".to_string(), "Broker unavailable".to_string()),
        ],
    };

    assert!(!result.all_succeeded());
    assert!(!result.any_succeeded());
}

// =========================================================================
// Filter Evaluation Tests
// =========================================================================

#[test]
fn test_get_json_pointer_value_simple() {
    use super::manager::get_json_pointer_value;

    let data = serde_json::json!({"id": "123", "name": "Test"});

    assert_eq!(get_json_pointer_value(&data, "/id"), Some(&serde_json::json!("123")));
    assert_eq!(get_json_pointer_value(&data, "/name"), Some(&serde_json::json!("Test")));
    assert_eq!(get_json_pointer_value(&data, "/missing"), None);
}

#[test]
fn test_get_json_pointer_value_nested() {
    use super::manager::get_json_pointer_value;

    let data = serde_json::json!({
        "user": {
            "profile": {
                "name": "Alice"
            }
        }
    });

    assert_eq!(
        get_json_pointer_value(&data, "/user/profile/name"),
        Some(&serde_json::json!("Alice"))
    );
}

#[test]
fn test_get_json_pointer_value_dot_notation() {
    use super::manager::get_json_pointer_value;

    let data = serde_json::json!({"user": {"name": "Bob"}});

    // Dot notation should be converted to JSON pointer
    assert_eq!(get_json_pointer_value(&data, "user.name"), Some(&serde_json::json!("Bob")));
}

#[test]
fn test_filter_condition_eq() {
    use super::manager::evaluate_filter_condition;
    use crate::schema::FilterOperator;

    assert!(evaluate_filter_condition(
        Some(&serde_json::json!("active")),
        FilterOperator::Eq,
        &serde_json::json!("active")
    ));

    assert!(!evaluate_filter_condition(
        Some(&serde_json::json!("active")),
        FilterOperator::Eq,
        &serde_json::json!("inactive")
    ));
}

#[test]
fn test_filter_condition_ne() {
    use super::manager::evaluate_filter_condition;
    use crate::schema::FilterOperator;

    assert!(evaluate_filter_condition(
        Some(&serde_json::json!("active")),
        FilterOperator::Ne,
        &serde_json::json!("inactive")
    ));

    assert!(!evaluate_filter_condition(
        Some(&serde_json::json!("active")),
        FilterOperator::Ne,
        &serde_json::json!("active")
    ));
}

#[test]
fn test_filter_condition_numeric_comparisons() {
    use super::manager::evaluate_filter_condition;
    use crate::schema::FilterOperator;

    // Greater than
    assert!(evaluate_filter_condition(
        Some(&serde_json::json!(100)),
        FilterOperator::Gt,
        &serde_json::json!(50)
    ));
    assert!(!evaluate_filter_condition(
        Some(&serde_json::json!(50)),
        FilterOperator::Gt,
        &serde_json::json!(100)
    ));

    // Greater than or equal
    assert!(evaluate_filter_condition(
        Some(&serde_json::json!(100)),
        FilterOperator::Gte,
        &serde_json::json!(100)
    ));

    // Less than
    assert!(evaluate_filter_condition(
        Some(&serde_json::json!(50)),
        FilterOperator::Lt,
        &serde_json::json!(100)
    ));

    // Less than or equal
    assert!(evaluate_filter_condition(
        Some(&serde_json::json!(100)),
        FilterOperator::Lte,
        &serde_json::json!(100)
    ));
}

#[test]
fn test_filter_condition_string_comparisons() {
    use super::manager::evaluate_filter_condition;
    use crate::schema::FilterOperator;

    // Contains
    assert!(evaluate_filter_condition(
        Some(&serde_json::json!("hello world")),
        FilterOperator::Contains,
        &serde_json::json!("world")
    ));

    // StartsWith
    assert!(evaluate_filter_condition(
        Some(&serde_json::json!("hello world")),
        FilterOperator::StartsWith,
        &serde_json::json!("hello")
    ));

    // EndsWith
    assert!(evaluate_filter_condition(
        Some(&serde_json::json!("hello world")),
        FilterOperator::EndsWith,
        &serde_json::json!("world")
    ));
}

#[test]
fn test_filter_condition_array_contains() {
    use super::manager::evaluate_filter_condition;
    use crate::schema::FilterOperator;

    assert!(evaluate_filter_condition(
        Some(&serde_json::json!(["a", "b", "c"])),
        FilterOperator::Contains,
        &serde_json::json!("b")
    ));

    assert!(!evaluate_filter_condition(
        Some(&serde_json::json!(["a", "b", "c"])),
        FilterOperator::Contains,
        &serde_json::json!("d")
    ));
}

#[test]
fn test_filter_condition_null_handling() {
    use super::manager::evaluate_filter_condition;
    use crate::schema::FilterOperator;

    // Missing value equals null
    assert!(evaluate_filter_condition(None, FilterOperator::Eq, &serde_json::Value::Null));

    // Missing value does not equal non-null
    assert!(!evaluate_filter_condition(
        None,
        FilterOperator::Eq,
        &serde_json::json!("value")
    ));
}

#[test]
fn test_subscription_filter_matching() {
    use std::collections::HashMap;

    use crate::schema::{FilterOperator, StaticFilterCondition, SubscriptionFilter};

    let mut argument_paths = HashMap::new();
    argument_paths.insert("orderId".to_string(), "/id".to_string());

    let filter = SubscriptionFilter {
        argument_paths,
        static_filters: vec![StaticFilterCondition {
            path:     "/status".to_string(),
            operator: FilterOperator::Eq,
            value:    serde_json::json!("active"),
        }],
    };

    let schema = Arc::new(CompiledSchema {
        subscriptions: vec![
            SubscriptionDefinition::new("OrderUpdated", "Order")
                .with_topic("order_updated")
                .with_filter(filter),
        ],
        ..Default::default()
    });

    let manager = SubscriptionManager::new(schema);

    // Subscribe with a specific orderId
    manager
        .subscribe(
            "OrderUpdated",
            serde_json::json!({}),
            serde_json::json!({"orderId": "ord_123"}),
            "conn_1",
        )
        .unwrap();

    // Event matching the filter
    let matching_event = SubscriptionEvent::new(
        "Order",
        "ord_123",
        SubscriptionOperation::Update,
        serde_json::json!({"id": "ord_123", "status": "active"}),
    );
    assert_eq!(manager.publish_event(matching_event), 1);

    // Event with wrong orderId
    let wrong_id_event = SubscriptionEvent::new(
        "Order",
        "ord_456",
        SubscriptionOperation::Update,
        serde_json::json!({"id": "ord_456", "status": "active"}),
    );
    assert_eq!(manager.publish_event(wrong_id_event), 0);

    // Event with wrong status
    let wrong_status_event = SubscriptionEvent::new(
        "Order",
        "ord_123",
        SubscriptionOperation::Update,
        serde_json::json!({"id": "ord_123", "status": "inactive"}),
    );
    assert_eq!(manager.publish_event(wrong_status_event), 0);
}

#[test]
fn test_subscription_field_projection() {
    let schema = Arc::new(CompiledSchema {
        subscriptions: vec![
            SubscriptionDefinition::new("OrderCreated", "Order")
                .with_topic("order_created")
                .with_fields(vec!["id".to_string(), "total".to_string()]),
        ],
        ..Default::default()
    });

    let manager = SubscriptionManager::new(schema);

    manager
        .subscribe("OrderCreated", serde_json::json!({}), serde_json::json!({}), "conn_1")
        .unwrap();

    let mut receiver = manager.receiver();

    let event = SubscriptionEvent::new(
        "Order",
        "ord_123",
        SubscriptionOperation::Create,
        serde_json::json!({
            "id": "ord_123",
            "total": 99.99,
            "secret_field": "should_not_appear",
            "customer": "John"
        }),
    );

    manager.publish_event(event);

    if let Ok(payload) = receiver.try_recv() {
        // Only projected fields should be present
        assert_eq!(payload.data.get("id"), Some(&serde_json::json!("ord_123")));
        assert_eq!(payload.data.get("total"), Some(&serde_json::json!(99.99)));
        assert!(payload.data.get("secret_field").is_none());
        assert!(payload.data.get("customer").is_none());
    }
}

// =========================================================================
// filter_fields Expansion Tests
// =========================================================================

#[test]
fn test_filter_fields_auto_generates_argument_paths() {
    let mut def = SubscriptionDefinition::new("OrderCreated", "Order").with_topic("order_created");
    def.filter_fields = vec!["user_id".to_string(), "tenant_id".to_string()];

    let schema = Arc::new(CompiledSchema {
        subscriptions: vec![def],
        ..Default::default()
    });

    let manager = SubscriptionManager::new(schema);

    // Subscribe with filter_fields variables
    let id = manager
        .subscribe(
            "OrderCreated",
            serde_json::json!({}),
            serde_json::json!({"user_id": "usr_1", "tenant_id": "t_1"}),
            "conn_1",
        )
        .unwrap();

    // The definition should now have argument_paths auto-generated
    let sub = manager.get_subscription(id).unwrap();
    let filter = sub.definition.filter.as_ref().expect("filter should exist");
    assert_eq!(filter.argument_paths.get("user_id"), Some(&"/user_id".to_string()));
    assert_eq!(filter.argument_paths.get("tenant_id"), Some(&"/tenant_id".to_string()));
}

#[test]
fn test_filter_fields_does_not_overwrite_explicit_argument_paths() {
    use std::collections::HashMap;

    use crate::schema::SubscriptionFilter;

    let mut argument_paths = HashMap::new();
    argument_paths.insert("user_id".to_string(), "/author/id".to_string());

    let mut def = SubscriptionDefinition::new("OrderCreated", "Order")
        .with_topic("order_created")
        .with_filter(SubscriptionFilter {
            argument_paths,
            static_filters: Vec::new(),
        });
    // filter_fields includes user_id (already in argument_paths) and tenant_id (new)
    def.filter_fields = vec!["user_id".to_string(), "tenant_id".to_string()];

    let schema = Arc::new(CompiledSchema {
        subscriptions: vec![def],
        ..Default::default()
    });

    let manager = SubscriptionManager::new(schema);

    let id = manager
        .subscribe(
            "OrderCreated",
            serde_json::json!({}),
            serde_json::json!({"user_id": "usr_1", "tenant_id": "t_1"}),
            "conn_1",
        )
        .unwrap();

    let sub = manager.get_subscription(id).unwrap();
    let filter = sub.definition.filter.as_ref().unwrap();
    // Explicit path should be preserved, not overwritten
    assert_eq!(filter.argument_paths.get("user_id"), Some(&"/author/id".to_string()));
    // New field should be auto-generated
    assert_eq!(filter.argument_paths.get("tenant_id"), Some(&"/tenant_id".to_string()));
}

#[test]
fn test_filter_fields_filtering_events() {
    let mut def = SubscriptionDefinition::new("OrderCreated", "Order").with_topic("order_created");
    def.filter_fields = vec!["user_id".to_string()];

    let schema = Arc::new(CompiledSchema {
        subscriptions: vec![def],
        ..Default::default()
    });

    let manager = SubscriptionManager::new(schema);

    // Subscribe with user_id filter
    manager
        .subscribe(
            "OrderCreated",
            serde_json::json!({}),
            serde_json::json!({"user_id": "usr_1"}),
            "conn_1",
        )
        .unwrap();

    // Matching event
    let matching = SubscriptionEvent::new(
        "Order",
        "ord_1",
        SubscriptionOperation::Create,
        serde_json::json!({"id": "ord_1", "user_id": "usr_1"}),
    );
    assert_eq!(manager.publish_event(matching), 1);

    // Non-matching event (different user_id)
    let non_matching = SubscriptionEvent::new(
        "Order",
        "ord_2",
        SubscriptionOperation::Create,
        serde_json::json!({"id": "ord_2", "user_id": "usr_2"}),
    );
    assert_eq!(manager.publish_event(non_matching), 0);
}

// =============================================================================
// RLS event filtering tests (C2 fix verification)
// =============================================================================

#[test]
fn test_rls_conditions_filter_subscription_events() {
    let schema = Arc::new(create_test_schema());
    let manager = SubscriptionManager::new(schema);

    // Subscribe with RLS: only tenant-abc events should match.
    manager
        .subscribe_with_rls(
            "OrderCreated",
            serde_json::json!({"user_id": "usr_1"}),
            serde_json::json!({}),
            "conn_1",
            vec![("tenant_id".to_string(), serde_json::json!("tenant-abc"))],
        )
        .unwrap();

    // Event for the correct tenant — should match.
    let matching = SubscriptionEvent::new(
        "Order",
        "ord_1",
        SubscriptionOperation::Create,
        serde_json::json!({"id": "ord_1", "tenant_id": "tenant-abc", "amount": 50}),
    );
    assert_eq!(manager.publish_event(matching), 1, "same-tenant event must match");

    // Event for a different tenant — must NOT match.
    let other_tenant = SubscriptionEvent::new(
        "Order",
        "ord_2",
        SubscriptionOperation::Create,
        serde_json::json!({"id": "ord_2", "tenant_id": "tenant-xyz", "amount": 75}),
    );
    assert_eq!(manager.publish_event(other_tenant), 0, "other-tenant event must be filtered");

    // Event with no tenant_id field — must NOT match.
    let missing_field = SubscriptionEvent::new(
        "Order",
        "ord_3",
        SubscriptionOperation::Create,
        serde_json::json!({"id": "ord_3", "amount": 100}),
    );
    assert_eq!(
        manager.publish_event(missing_field),
        0,
        "event without tenant_id must be filtered"
    );
}

// =============================================================================
// Tenant-gate fail-closed tests (C7 — M-tenant-ws-failopen)
// =============================================================================

#[test]
fn multi_tenant_tenantless_subscriber_receives_no_events() {
    // The fail-open hole: a subscriber whose context carries NO tenant_id used to
    // match every tenant's events. In multi-tenant mode it must now match nothing.
    let manager = SubscriptionManager::new(Arc::new(create_multi_tenant_schema()));
    manager
        .subscribe(
            "OrderCreated",
            serde_json::json!({ "user_id": "u1" }), // no tenant_id
            serde_json::json!({}),
            "conn_1",
        )
        .unwrap();

    let event = SubscriptionEvent::new(
        "Order",
        "ord_1",
        SubscriptionOperation::Create,
        serde_json::json!({ "id": "ord_1" }),
    )
    .with_tenant_id("tenant-a");

    assert_eq!(
        manager.publish_event(event),
        0,
        "multi-tenant: a tenant-less subscriber must not receive any tenant's events",
    );
}

#[test]
fn multi_tenant_subscriber_receives_only_same_tenant_events() {
    let manager = SubscriptionManager::new(Arc::new(create_multi_tenant_schema()));
    manager
        .subscribe(
            "OrderCreated",
            serde_json::json!({ "tenant_id": "tenant-a" }),
            serde_json::json!({}),
            "conn_1",
        )
        .unwrap();

    let same = SubscriptionEvent::new(
        "Order",
        "ord_1",
        SubscriptionOperation::Create,
        serde_json::json!({ "id": "ord_1" }),
    )
    .with_tenant_id("tenant-a");
    assert_eq!(manager.publish_event(same), 1, "same-tenant event must match");

    let other = SubscriptionEvent::new(
        "Order",
        "ord_2",
        SubscriptionOperation::Create,
        serde_json::json!({ "id": "ord_2" }),
    )
    .with_tenant_id("tenant-b");
    assert_eq!(manager.publish_event(other), 0, "different-tenant event must be filtered");

    // An event with no tenant tag must NOT reach a tenant-scoped subscriber in
    // multi-tenant mode (it could be from any tenant).
    let untagged = SubscriptionEvent::new(
        "Order",
        "ord_3",
        SubscriptionOperation::Create,
        serde_json::json!({ "id": "ord_3" }),
    );
    assert_eq!(
        manager.publish_event(untagged),
        0,
        "multi-tenant: an untagged event must not reach a tenant-scoped subscriber",
    );
}

#[test]
fn single_tenant_tenantless_subscriber_still_matches() {
    // Regression guard: single-tenant deployments (tenant ids typically absent)
    // must keep delivering events to tenant-less subscribers.
    let manager = SubscriptionManager::new(Arc::new(create_test_schema()));
    manager
        .subscribe(
            "OrderCreated",
            serde_json::json!({ "user_id": "u1" }),
            serde_json::json!({}),
            "conn_1",
        )
        .unwrap();

    let event = SubscriptionEvent::new(
        "Order",
        "ord_1",
        SubscriptionOperation::Create,
        serde_json::json!({ "id": "ord_1" }),
    );
    assert_eq!(
        manager.publish_event(event),
        1,
        "single-tenant: a tenant-less subscriber must still receive untagged events",
    );
}

#[test]
fn test_rls_conditions_multiple_fields() {
    let schema = Arc::new(create_test_schema());
    let manager = SubscriptionManager::new(schema);

    // Subscribe with RLS: tenant + author must both match.
    manager
        .subscribe_with_rls(
            "OrderCreated",
            serde_json::json!({}),
            serde_json::json!({}),
            "conn_1",
            vec![
                ("tenant_id".to_string(), serde_json::json!("tenant-abc")),
                ("author_id".to_string(), serde_json::json!("user-1")),
            ],
        )
        .unwrap();

    // Both match.
    let both_match = SubscriptionEvent::new(
        "Order",
        "ord_1",
        SubscriptionOperation::Create,
        serde_json::json!({"id": "ord_1", "tenant_id": "tenant-abc", "author_id": "user-1"}),
    );
    assert_eq!(manager.publish_event(both_match), 1);

    // Tenant matches but author doesn't.
    let wrong_author = SubscriptionEvent::new(
        "Order",
        "ord_2",
        SubscriptionOperation::Create,
        serde_json::json!({"id": "ord_2", "tenant_id": "tenant-abc", "author_id": "user-2"}),
    );
    assert_eq!(manager.publish_event(wrong_author), 0);
}

#[test]
fn test_rls_empty_conditions_allows_all() {
    let schema = Arc::new(create_test_schema());
    let manager = SubscriptionManager::new(schema);

    // Admin: no RLS conditions.
    manager
        .subscribe_with_rls(
            "OrderCreated",
            serde_json::json!({}),
            serde_json::json!({}),
            "conn_admin",
            Vec::new(),
        )
        .unwrap();

    let event = SubscriptionEvent::new(
        "Order",
        "ord_1",
        SubscriptionOperation::Create,
        serde_json::json!({"id": "ord_1", "tenant_id": "any-tenant"}),
    );
    assert_eq!(manager.publish_event(event), 1, "admin with no RLS should see all events");
}

#[test]
fn test_extract_rls_conditions_from_where_clause() {
    use super::super::extract_rls_conditions;
    use crate::db::{WhereClause, WhereOperator};

    // AND(tenant_id = "abc", author_id = "user-1")
    let clause = WhereClause::And(vec![
        WhereClause::Field {
            path:     vec!["tenant_id".to_string()],
            operator: WhereOperator::Eq,
            value:    serde_json::json!("abc"),
        },
        WhereClause::Field {
            path:     vec!["author_id".to_string()],
            operator: WhereOperator::Eq,
            value:    serde_json::json!("user-1"),
        },
    ]);

    let conditions = extract_rls_conditions(&clause);
    assert_eq!(conditions.len(), 2);
    assert_eq!(conditions[0], ("tenant_id".to_string(), serde_json::json!("abc")));
    assert_eq!(conditions[1], ("author_id".to_string(), serde_json::json!("user-1")));
}

#[test]
fn test_extract_rls_conditions_ignores_non_eq() {
    use super::super::extract_rls_conditions;
    use crate::db::{WhereClause, WhereOperator};

    // Only Eq conditions are extracted; Gt is skipped.
    let clause = WhereClause::And(vec![
        WhereClause::Field {
            path:     vec!["tenant_id".to_string()],
            operator: WhereOperator::Eq,
            value:    serde_json::json!("abc"),
        },
        WhereClause::Field {
            path:     vec!["score".to_string()],
            operator: WhereOperator::Gt,
            value:    serde_json::json!(100),
        },
    ]);

    let conditions = extract_rls_conditions(&clause);
    assert_eq!(conditions.len(), 1, "only Eq conditions should be extracted");
    assert_eq!(conditions[0].0, "tenant_id");
}

// ── C20: Subscription RBAC error variants ────────────────────────────────

/// C20: Verify `AuthenticationRequired` and `Forbidden` error variants exist
/// and can be constructed. These are reserved for future subscription-level
/// RBAC enforcement (e.g., `requires_role` on subscription definitions).
///
/// Current state: the `subscribe()` method does not check authentication or
/// authorization — it accepts any `user_context`. These variants are not dead
/// code but rather forward-looking API surface for when subscription RBAC is
/// implemented.
#[test]
fn test_subscription_error_rbac_variants_constructible() {
    let auth_err = SubscriptionError::AuthenticationRequired("onOrderCreated".to_string());
    assert!(
        format!("{auth_err}").contains("Authentication required"),
        "AuthenticationRequired variant should format correctly: {auth_err}"
    );

    let forbidden_err = SubscriptionError::Forbidden("onOrderCreated".to_string());
    assert!(
        format!("{forbidden_err}").contains("Not authorized"),
        "Forbidden variant should format correctly: {forbidden_err}"
    );
}

mod change_spine_envelope_tests {
    #![allow(clippy::unwrap_used)]
    use super::super::{ChangeSpineEnvelope, SubscriptionEvent, SubscriptionOperation};

    #[test]
    fn serializes_camel_case_and_omits_none_fields() {
        let env = ChangeSpineEnvelope {
            actor_type:     Some("ai_agent".to_string()),
            acting_for:     None,
            schema_version: Some("v3".to_string()),
            tenant_id:      None,
            duration_ms:    Some(12),
            seq:            Some(42),
        };
        let v = serde_json::to_value(&env).unwrap();
        assert_eq!(v["actorType"], "ai_agent");
        assert_eq!(v["schemaVersion"], "v3");
        assert_eq!(v["durationMs"], 12);
        assert_eq!(v["seq"], 42);
        // None fields are omitted entirely (not serialized as null).
        let obj = v.as_object().unwrap();
        assert!(!obj.contains_key("actingFor"), "None acting_for is omitted");
        assert!(!obj.contains_key("tenantId"), "None tenant_id is omitted");
    }

    #[test]
    fn is_empty_distinguishes_stamped_from_unstamped() {
        assert!(ChangeSpineEnvelope::default().is_empty());
        let env = ChangeSpineEnvelope {
            actor_type: Some("system_job".to_string()),
            ..Default::default()
        };
        assert!(!env.is_empty());
    }

    #[test]
    fn with_change_spine_attaches_envelope_to_event() {
        let env = ChangeSpineEnvelope {
            actor_type: Some("human_user".to_string()),
            ..Default::default()
        };
        let event = SubscriptionEvent::new(
            "Order",
            "ord_1",
            SubscriptionOperation::Update,
            serde_json::json!({"id": "ord_1"}),
        )
        .with_change_spine(env);
        assert_eq!(
            event.change_spine.expect("envelope attached").actor_type.as_deref(),
            Some("human_user")
        );
    }
}

mod protocol_tests {
    #![allow(clippy::unwrap_used)]
    use super::super::protocol::*;

    #[test]
    fn test_client_message_type_parsing() {
        assert_eq!(
            ClientMessageType::from_str("connection_init"),
            Some(ClientMessageType::ConnectionInit)
        );
        assert_eq!(ClientMessageType::from_str("subscribe"), Some(ClientMessageType::Subscribe));
        assert_eq!(ClientMessageType::from_str("invalid"), None);
    }

    #[test]
    fn test_server_message_connection_ack() {
        let msg = ServerMessage::connection_ack(None);
        assert_eq!(msg.message_type, "connection_ack");
        assert!(msg.id.is_none());

        let json = msg.to_json().unwrap();
        assert!(json.contains("connection_ack"));
    }

    #[test]
    fn test_server_message_next() {
        let data = serde_json::json!({"orderCreated": {"id": "ord_123"}});
        let msg = ServerMessage::next("op_1", data);

        assert_eq!(msg.message_type, "next");
        assert_eq!(msg.id, Some("op_1".to_string()));

        let json = msg.to_json().unwrap();
        assert!(json.contains("next"));
        assert!(json.contains("op_1"));
        assert!(json.contains("orderCreated"));
    }

    #[test]
    fn next_with_extensions_carries_data_and_extensions() {
        // The next payload must be the graphql-transport-ws ExecutionResult shape:
        // { data, extensions } — data untouched, envelope under extensions (#425).
        let data = serde_json::json!({ "orderUpdated": { "id": "ord_7" } });
        let extensions = serde_json::json!({ "changeSpine": { "actorType": "human_user" } });
        let msg = ServerMessage::next_with_extensions("op_2", data, extensions);

        assert_eq!(msg.message_type, "next");
        assert_eq!(msg.id, Some("op_2".to_string()));
        let payload = msg.payload.expect("next message has a payload");
        assert_eq!(payload["data"]["orderUpdated"]["id"], "ord_7");
        assert_eq!(payload["extensions"]["changeSpine"]["actorType"], "human_user");
    }

    #[test]
    fn test_server_message_error() {
        let errors = vec![GraphQLError::with_code(
            "Subscription not found",
            "SUBSCRIPTION_NOT_FOUND",
        )];
        let msg = ServerMessage::error("op_1", errors);

        assert_eq!(msg.message_type, "error");
        let json = msg.to_json().unwrap();
        assert!(json.contains("Subscription not found"));
    }

    #[test]
    fn test_server_message_complete() {
        let msg = ServerMessage::complete("op_1");

        assert_eq!(msg.message_type, "complete");
        assert_eq!(msg.id, Some("op_1".to_string()));
        assert!(msg.payload.is_none());
    }

    #[test]
    fn test_client_message_parsing() {
        let json = r#"{
            "type": "subscribe",
            "id": "op_1",
            "payload": {
                "query": "subscription { orderCreated { id } }"
            }
        }"#;

        let msg: ClientMessage = serde_json::from_str(json).unwrap();
        assert_eq!(msg.parsed_type(), Some(ClientMessageType::Subscribe));
        assert_eq!(msg.id, Some("op_1".to_string()));

        let payload = msg.subscription_payload().unwrap();
        assert!(payload.query.contains("orderCreated"));
    }

    #[test]
    fn test_close_codes() {
        assert_eq!(CloseCode::Normal.code(), 1000);
        assert_eq!(CloseCode::Unauthorized.code(), 4401);
        assert_eq!(CloseCode::SubscriberAlreadyExists.code(), 4409);
    }

    #[test]
    fn test_graphql_error() {
        let error = GraphQLError::with_code("Test error", "TEST_ERROR");
        assert_eq!(error.message, "Test error");
        assert!(error.extensions.is_some());

        let json = serde_json::to_string(&error).unwrap();
        assert!(json.contains("TEST_ERROR"));
    }
}
