aingle_cortex 0.7.5

// Copyright 2019-2026 Apilium Technologies OÜ. All rights reserved.
// SPDX-License-Identifier: Apache-2.0 OR Commercial

//! Integration tests for GraphQL subscriptions
//!
//! Tests WebSocket-based real-time updates for:
//! - Triple additions/deletions
//! - Validation events
//! - Agent activity
//! - Heartbeats

#[cfg(feature = "graphql")]
mod tests {
    use aingle_cortex::{AppState, CortexConfig, CortexServer};
    use aingle_graph::{GraphDB, NodeId, Predicate, Triple, Value};
    use futures::StreamExt;
    use std::time::Duration;
    use tokio::time::timeout;

    /// Helper to create test server
    fn create_test_server() -> CortexServer {
        let config = CortexConfig::default().with_host("127.0.0.1").with_port(0); // Random port

        CortexServer::new(config).expect("Failed to create server")
    }

    /// Helper to add a test triple
    async fn add_test_triple(state: &AppState, subject: &str, predicate: &str, object: &str) {
        let mut graph = state.graph.write().await;
        let triple = Triple::new(
            NodeId::named(subject),
            Predicate::named(predicate),
            Value::String(object.to_string()),
        );
        graph.insert(triple).expect("Failed to add triple");
    }

    /// Helper to broadcast an event
    fn broadcast_event(state: &AppState, event: aingle_cortex::state::Event) {
        state.broadcaster.broadcast(event);
    }

    #[tokio::test]
    async fn test_triple_added_subscription() {
        let server = create_test_server();
        let state = server.state().clone();

        // Subscribe to events
        let mut rx = state.broadcaster.subscribe();

        // Add a triple (this will trigger an event if broadcaster is used in real implementation)
        broadcast_event(
            &state,
            aingle_cortex::state::Event::TripleAdded {
                hash: "test_hash_123".to_string(),
                subject: "ex:Alice".to_string(),
                predicate: "foaf:name".to_string(),
                object: serde_json::Value::String("Alice".to_string()),
            },
        );

        // Receive event
        let result = timeout(Duration::from_secs(1), rx.recv()).await;
        assert!(result.is_ok(), "Should receive event within timeout");

        let event = result.unwrap().unwrap();
        match event {
            aingle_cortex::state::Event::TripleAdded {
                hash,
                subject,
                predicate,
                ..
            } => {
                assert_eq!(hash, "test_hash_123");
                assert_eq!(subject, "ex:Alice");
                assert_eq!(predicate, "foaf:name");
            }
            _ => panic!("Expected TripleAdded event"),
        }
    }

    #[tokio::test]
    async fn test_triple_deleted_subscription() {
        let server = create_test_server();
        let state = server.state().clone();

        let mut rx = state.broadcaster.subscribe();

        // Broadcast delete event
        broadcast_event(
            &state,
            aingle_cortex::state::Event::TripleDeleted {
                hash: "deleted_hash".to_string(),
            },
        );

        // Receive event
        let result = timeout(Duration::from_secs(1), rx.recv()).await;
        assert!(result.is_ok());

        let event = result.unwrap().unwrap();
        match event {
            aingle_cortex::state::Event::TripleDeleted { hash } => {
                assert_eq!(hash, "deleted_hash");
            }
            _ => panic!("Expected TripleDeleted event"),
        }
    }

    #[tokio::test]
    async fn test_validation_event_subscription() {
        let server = create_test_server();
        let state = server.state().clone();

        let mut rx = state.broadcaster.subscribe();

        // Broadcast validation event
        broadcast_event(
            &state,
            aingle_cortex::state::Event::ValidationCompleted {
                hash: "validated_hash".to_string(),
                valid: true,
                proof_hash: Some("proof_123".to_string()),
            },
        );

        // Receive event
        let result = timeout(Duration::from_secs(1), rx.recv()).await;
        assert!(result.is_ok());

        let event = result.unwrap().unwrap();
        match event {
            aingle_cortex::state::Event::ValidationCompleted {
                hash,
                valid,
                proof_hash,
            } => {
                assert_eq!(hash, "validated_hash");
                assert!(valid);
                assert_eq!(proof_hash, Some("proof_123".to_string()));
            }
            _ => panic!("Expected ValidationCompleted event"),
        }
    }

    #[tokio::test]
    async fn test_multiple_subscribers() {
        let server = create_test_server();
        let state = server.state().clone();

        // Create multiple subscribers
        let mut rx1 = state.broadcaster.subscribe();
        let mut rx2 = state.broadcaster.subscribe();
        let mut rx3 = state.broadcaster.subscribe();

        // Broadcast event
        broadcast_event(
            &state,
            aingle_cortex::state::Event::TripleAdded {
                hash: "multi_hash".to_string(),
                subject: "ex:Test".to_string(),
                predicate: "ex:test".to_string(),
                object: serde_json::Value::String("test".to_string()),
            },
        );

        // All subscribers should receive the event
        let result1 = timeout(Duration::from_secs(1), rx1.recv()).await;
        let result2 = timeout(Duration::from_secs(1), rx2.recv()).await;
        let result3 = timeout(Duration::from_secs(1), rx3.recv()).await;

        assert!(result1.is_ok());
        assert!(result2.is_ok());
        assert!(result3.is_ok());
    }

    #[tokio::test]
    async fn test_client_count() {
        let server = create_test_server();
        let state = server.state().clone();

        // Initial count should be 0
        assert_eq!(state.broadcaster.client_count(), 0);

        // Subscribe
        let _rx1 = state.broadcaster.subscribe();
        assert_eq!(state.broadcaster.client_count(), 1);

        let _rx2 = state.broadcaster.subscribe();
        assert_eq!(state.broadcaster.client_count(), 2);

        // Drop subscribers
        drop(_rx1);
        drop(_rx2);

        // Note: client_count doesn't auto-decrement on drop in current implementation
        // This would need to be implemented with a custom Drop guard
    }

    #[tokio::test]
    async fn test_event_filtering() {
        let server = create_test_server();
        let state = server.state().clone();

        let mut rx = state.broadcaster.subscribe();

        // Broadcast multiple events
        broadcast_event(
            &state,
            aingle_cortex::state::Event::TripleAdded {
                hash: "hash1".to_string(),
                subject: "ex:Alice".to_string(),
                predicate: "foaf:name".to_string(),
                object: serde_json::Value::String("Alice".to_string()),
            },
        );

        broadcast_event(
            &state,
            aingle_cortex::state::Event::TripleDeleted {
                hash: "hash2".to_string(),
            },
        );

        broadcast_event(
            &state,
            aingle_cortex::state::Event::ValidationCompleted {
                hash: "hash3".to_string(),
                valid: true,
                proof_hash: None,
            },
        );

        // Receive all events
        let mut received = Vec::new();
        for _ in 0..3 {
            if let Ok(Ok(event)) = timeout(Duration::from_millis(500), rx.recv()).await {
                received.push(event);
            }
        }

        assert_eq!(received.len(), 3);

        // Verify event types
        assert!(matches!(
            received[0],
            aingle_cortex::state::Event::TripleAdded { .. }
        ));
        assert!(matches!(
            received[1],
            aingle_cortex::state::Event::TripleDeleted { .. }
        ));
        assert!(matches!(
            received[2],
            aingle_cortex::state::Event::ValidationCompleted { .. }
        ));
    }

    #[tokio::test]
    async fn test_event_json_serialization() {
        let server = create_test_server();
        let state = server.state().clone();

        let event = aingle_cortex::state::Event::TripleAdded {
            hash: "json_hash".to_string(),
            subject: "ex:Subject".to_string(),
            predicate: "ex:predicate".to_string(),
            object: serde_json::json!({"value": "test"}),
        };

        // Convert to JSON
        let json = event.to_json();
        assert!(json.contains("TripleAdded"));
        assert!(json.contains("json_hash"));
        assert!(json.contains("ex:Subject"));

        // Verify it's valid JSON
        let parsed: serde_json::Value = serde_json::from_str(&json).expect("Invalid JSON");
        assert_eq!(parsed["type"], "TripleAdded");
    }

    #[tokio::test]
    async fn test_subscription_with_filter() {
        let server = create_test_server();
        let state = server.state().clone();

        let mut rx = state.broadcaster.subscribe();

        // Broadcast events with different predicates
        broadcast_event(
            &state,
            aingle_cortex::state::Event::TripleAdded {
                hash: "h1".to_string(),
                subject: "ex:Alice".to_string(),
                predicate: "foaf:knows".to_string(),
                object: serde_json::Value::String("Bob".to_string()),
            },
        );

        broadcast_event(
            &state,
            aingle_cortex::state::Event::TripleAdded {
                hash: "h2".to_string(),
                subject: "ex:Alice".to_string(),
                predicate: "foaf:name".to_string(),
                object: serde_json::Value::String("Alice".to_string()),
            },
        );

        // Manually filter for "foaf:knows"
        let mut foaf_knows_events = Vec::new();
        for _ in 0..2 {
            if let Ok(Ok(event)) = timeout(Duration::from_millis(500), rx.recv()).await {
                if let aingle_cortex::state::Event::TripleAdded { predicate, .. } = &event {
                    if predicate == "foaf:knows" {
                        foaf_knows_events.push(event);
                    }
                }
            }
        }

        assert_eq!(foaf_knows_events.len(), 1);
    }

    #[tokio::test]
    async fn test_connected_event() {
        let server = create_test_server();
        let state = server.state().clone();

        let mut rx = state.broadcaster.subscribe();

        broadcast_event(
            &state,
            aingle_cortex::state::Event::Connected {
                client_id: "client_123".to_string(),
            },
        );

        let result = timeout(Duration::from_secs(1), rx.recv()).await;
        assert!(result.is_ok());

        let event = result.unwrap().unwrap();
        match event {
            aingle_cortex::state::Event::Connected { client_id } => {
                assert_eq!(client_id, "client_123");
            }
            _ => panic!("Expected Connected event"),
        }
    }

    #[tokio::test]
    async fn test_ping_event() {
        let server = create_test_server();
        let state = server.state().clone();

        let mut rx = state.broadcaster.subscribe();

        broadcast_event(&state, aingle_cortex::state::Event::Ping);

        let result = timeout(Duration::from_secs(1), rx.recv()).await;
        assert!(result.is_ok());

        let event = result.unwrap().unwrap();
        assert!(matches!(event, aingle_cortex::state::Event::Ping));
    }

    #[tokio::test]
    async fn test_subscription_buffer_overflow() {
        let server = create_test_server();
        let state = server.state().clone();

        let _rx = state.broadcaster.subscribe();

        // Broadcast many events (more than buffer capacity)
        for i in 0..2000 {
            broadcast_event(
                &state,
                aingle_cortex::state::Event::TripleAdded {
                    hash: format!("hash_{}", i),
                    subject: "ex:Test".to_string(),
                    predicate: "ex:test".to_string(),
                    object: serde_json::Value::String("test".to_string()),
                },
            );
        }

        // This test verifies that broadcasting doesn't panic even if buffers overflow
        // (tokio broadcast channel drops old messages when full)
    }
}