firebase-rust-sdk 0.1.0-beta

//! Integration tests for Firestore
//!
//! TEST SUCCESS CRITERIA: ALL 31 tests MUST pass when run in parallel with --test-threads=31
//!
//! Requirements:
//! 1. Firebase project with Firestore enabled
//! 2. Environment variables set in .env file
//! 3. ALL tests must pass consistently (5 consecutive runs)
//!
//! Parallel Execution Requirements:
//! - ALL tests share a SINGLE Firestore instance (production pattern)
//! - ALL 31 tests run concurrently (--test-threads=31)
//! - Zero random delays - tests must handle real concurrent load
//! - Unique collection names prevent test data conflicts
//! - Connection must be ready immediately (use connect() not connect_lazy())
//!
//! Acceptance:
//! - Run: cargo test --test firestore_integration -- --test-threads=31
//! - Result: test result: ok. 31 passed; 0 failed
//! - Must pass 5 consecutive times without failures
//!
//! Note: Uses gRPC API. Tests real-world concurrent access pattern.

use firebase_rust_sdk::{
    firestore::{
        listen_document, FilterCondition, Firestore, ListenerOptions, MapValue, Value, ValueType,
    },
    App, AppOptions, Auth,
};
use futures::stream::StreamExt;
use once_cell::sync::Lazy;
use rand::Rng;
use std::collections::HashMap;
use std::env;
use std::sync::Arc;
use tokio::runtime::Runtime;
use tokio::sync::{Mutex, OnceCell};
use tracing::{debug, error, info, warn};

/// Shared tokio runtime - CRITICAL for preventing tower buffer worker cancellation!
/// Problem: #[test] creates per-test runtimes that drop after each test
/// Solution: One shared multi-threaded runtime keeps background tasks (tower buffer worker) alive
/// Using multi-threaded runtime to support parallel test execution with --test-threads=31
static SHARED_RUNTIME: Lazy<Runtime> = Lazy::new(|| {
    tokio::runtime::Builder::new_multi_thread()
        .worker_threads(8) // Enough threads to handle 31 concurrent tests
        .enable_all()
        .build()
        .expect("Failed to create shared runtime")
});

/// Shared Firestore instance - initialized once, shared by ALL tests
/// This tests production usage pattern (single instance, concurrent access)
static SHARED_FIRESTORE: OnceCell<Arc<Firestore>> = OnceCell::const_new();

/// Initialize shared Firestore instance - called once, result cached and shared
async fn init_firestore() -> Arc<Firestore> {
    dotenvy::dotenv().ok();

    let project_id =
        env::var("FIREBASE_PROJECT_ID").expect("FIREBASE_PROJECT_ID must be set in .env file");
    let database_id = env::var("FIREBASE_DATABASE_ID").unwrap_or_else(|_| "default".to_string());
    let api_key = env::var("FIREBASE_API_KEY").expect("FIREBASE_API_KEY must be set in .env file");
    let email = env::var("TEST_USER_EMAIL").expect("TEST_USER_EMAIL must be set in .env file");
    let password =
        env::var("TEST_USER_PASSWORD").expect("TEST_USER_PASSWORD must be set in .env file");

    // Create App and Auth instances
    let app = App::create(AppOptions {
        api_key: api_key.clone(),
        project_id: project_id.clone(),
        app_name: None,
    })
    .await
    .expect("Failed to create App");

    let auth = Auth::get_auth(&app)
        .await
        .expect("Failed to get Auth instance");

    // Sign in to get ID token
    auth.sign_in_with_email_and_password(&email, &password)
        .await
        .expect("Failed to sign in - check TEST_USER_EMAIL and TEST_USER_PASSWORD");

    let user = auth
        .current_user()
        .await
        .expect("No current user after sign in");
    let id_token = user
        .get_id_token(false)
        .await
        .expect("Failed to get ID token");

    let firestore = Firestore::new(project_id, database_id, Some(id_token))
        .await
        .expect("Failed to create Firestore instance");

    Arc::new(firestore)
}

/// Get the SHARED Firestore instance (all tests use same instance)
/// This tests production usage: single instance handling concurrent operations
/// Returns &'static reference to keep the shared instance alive for entire test run
async fn get_firestore() -> &'static Arc<Firestore> {
    // Initialize tracing once
    // static TRACING_INIT: OnceCell<()> = OnceCell::const_new();
    // TRACING_INIT
    //     .get_or_init(|| async {
    //         tracing_subscriber::fmt()
    //             .with_env_filter(
    //                 tracing_subscriber::EnvFilter::from_default_env()
    //                     .add_directive(tracing::Level::DEBUG.into()),
    //             )
    //             .with_target(true)
    //             .with_thread_ids(true)
    //             .with_line_number(true)
    //             .init();
    //         info!("Tracing initialized");
    //     })
    //     .await;

    // Return reference to the OnceCell's Arc - keeps it alive for entire program
    SHARED_FIRESTORE.get_or_init(|| init_firestore()).await
}

/// Helper to create a MapValue from key-value pairs
fn create_map(fields: Vec<(&str, ValueType)>) -> MapValue {
    let mut map = HashMap::new();
    for (key, value_type) in fields {
        map.insert(
            key.to_string(),
            Value {
                value_type: Some(value_type),
            },
        );
    }
    MapValue { fields: map }
}

/// Generate unique document path for testing
fn test_doc_path(test_name: &str) -> String {
    let timestamp = chrono::Utc::now().timestamp();
    let random = rand::random::<u32>();
    format!("integration_tests/{}_{}_{}", test_name, timestamp, random)
}

/// Test: Create and read document using gRPC
#[test]
fn test_set_and_get_document() {
    SHARED_RUNTIME.block_on(async {
        let firestore = get_firestore().await;
        let doc_path = test_doc_path("set_get");

        // Create document using DocumentReference.set()
        let data = create_map(vec![
            ("name", ValueType::StringValue("Alice".to_string())),
            ("age", ValueType::IntegerValue(30)),
            ("active", ValueType::BooleanValue(true)),
        ]);

        let doc_ref = firestore.document(&doc_path);
        doc_ref.set(data).await.expect("Failed to set document");

        // Read document using DocumentReference.get()
        let snapshot = doc_ref.get().await.expect("Failed to get document");

        assert!(snapshot.exists());

        // Verify field values
        let name = snapshot.get("name").expect("name field missing");
        match &name.value_type {
            Some(ValueType::StringValue(s)) => assert_eq!(s, "Alice"),
            _ => panic!("Expected string value for name"),
        }

        let age = snapshot.get("age").expect("age field missing");
        match age.value_type {
            Some(ValueType::IntegerValue(i)) => assert_eq!(i, 30),
            _ => panic!("Expected integer value for age"),
        }

        // Clean up
        doc_ref.delete().await.expect("Failed to delete document");

        println!("✅ Set and get document works!");
    });
}

/// Test: Update document using gRPC
#[test]
fn test_update_document() {
    SHARED_RUNTIME.block_on(async {
        let firestore = get_firestore().await;
        let doc_path = test_doc_path("update");

        let doc_ref = firestore.document(&doc_path);

        // Create document with initial data
        let initial_data = create_map(vec![
            ("count", ValueType::IntegerValue(0)),
            ("name", ValueType::StringValue("Test".to_string())),
        ]);
        doc_ref
            .set(initial_data)
            .await
            .expect("Failed to create document");

        // Update only count field (should keep name)
        let update_data = create_map(vec![("count", ValueType::IntegerValue(42))]);
        doc_ref
            .update(update_data)
            .await
            .expect("Failed to update document");

        // Read updated document
        let snapshot = doc_ref.get().await.expect("Failed to get document");

        // Verify count was updated
        let count = snapshot.get("count").expect("count field missing");
        match count.value_type {
            Some(ValueType::IntegerValue(i)) => assert_eq!(i, 42, "Count should be updated to 42"),
            _ => panic!("Expected integer value for count"),
        }

        // Verify name still exists with original value (update doesn't replace entire document)
        let name = snapshot
            .get("name")
            .expect("name field should still exist after update");
        match &name.value_type {
            Some(ValueType::StringValue(s)) => {
                assert_eq!(s, "Test", "Name should still be 'Test' after update")
            }
            _ => panic!("Expected string value for name"),
        }

        // Clean up
        doc_ref.delete().await.expect("Failed to delete document");

        println!("✅ Update document works!");
    });
}

/// Test: Delete document using gRPC
#[test]
fn test_delete_document() {
    SHARED_RUNTIME.block_on(async {
        let firestore = get_firestore().await;
        let doc_path = test_doc_path("delete");

        let doc_ref = firestore.document(&doc_path);

        // Create document
        let data = create_map(vec![("test", ValueType::BooleanValue(true))]);
        doc_ref.set(data).await.expect("Failed to create document");

        // Verify it exists
        let snapshot = doc_ref.get().await.expect("Failed to get document");
        assert!(snapshot.exists());

        // Delete document
        doc_ref.delete().await.expect("Failed to delete document");

        // Verify it's gone - Firestore returns NotFound error for deleted documents
        let result = doc_ref.get().await;
        assert!(result.is_err() || !result.unwrap().exists());

        println!("✅ Delete document works!");
    });
}

/// Test: WriteBatch with multiple operations using gRPC
#[test]
fn test_write_batch() {
    SHARED_RUNTIME.block_on(async {
        let firestore = get_firestore().await;
        let collection_path = format!("integration_tests_batch_{}", rand::random::<u32>());

        // Create batch
        let mut batch = firestore.batch();

        // Add multiple write operations
        for i in 1..=3 {
            let doc_path = format!("{}/doc{}", collection_path, i);
            let data = create_map(vec![
                ("index", ValueType::IntegerValue(i)),
                ("batch_test", ValueType::BooleanValue(true)),
            ]);
            batch = batch.set(doc_path, data);
        }

        // Commit batch (atomic - all succeed or all fail)
        batch.commit().await.expect("Failed to commit batch");

        // Verify all documents exist
        for i in 1..=3 {
            let doc_path = format!("{}/doc{}", collection_path, i);
            let doc_ref = firestore.document(&doc_path);
            let snapshot = doc_ref.get().await.expect("Failed to read document");

            assert!(snapshot.exists());

            let index = snapshot.get("index").expect("index field missing");
            match index.value_type {
                Some(ValueType::IntegerValue(val)) => assert_eq!(val, i),
                _ => panic!("Expected integer value for index"),
            }

            // Clean up
            doc_ref.delete().await.expect("Failed to delete");
        }

        println!("✅ WriteBatch works!");
    });
}

/// Test: CollectionReference.add() with auto-generated ID
#[test]
fn test_collection_add() {
    SHARED_RUNTIME.block_on(async {
        let firestore = get_firestore().await;
        let collection_path = format!("integration_tests_add_{}", rand::random::<u32>());

        let collection_ref = firestore.collection(&collection_path);

        // Add document with auto-generated ID
        let data = create_map(vec![
            (
                "message",
                ValueType::StringValue("Auto-generated ID".to_string()),
            ),
            (
                "timestamp",
                ValueType::IntegerValue(chrono::Utc::now().timestamp()),
            ),
        ]);

        let doc_ref = collection_ref
            .add(data)
            .await
            .expect("Failed to add document");

        // Verify document was created
        assert!(doc_ref.path.starts_with(&collection_path));
        assert_eq!(doc_ref.id().len(), 20); // Auto-generated IDs are 20 chars

        let snapshot = doc_ref.get().await.expect("Failed to read document");

        assert!(snapshot.exists());

        let message = snapshot.get("message").expect("message field missing");
        match &message.value_type {
            Some(ValueType::StringValue(s)) => assert_eq!(s, "Auto-generated ID"),
            _ => panic!("Expected string value for message"),
        }

        // Clean up
        doc_ref.delete().await.expect("Failed to delete document");

        println!("✅ CollectionReference.add() works!");
    });
}

/// Test: CollectionReference.document() path parsing
#[test]
fn test_collection_document_reference() {
    SHARED_RUNTIME.block_on(async {
        let firestore = get_firestore().await;

        let collection_ref = firestore.collection("users");
        let doc_ref = collection_ref.document("alice");

        assert_eq!(doc_ref.path, "users/alice");
        assert_eq!(doc_ref.id(), "alice");
        assert_eq!(doc_ref.parent_path(), Some("users"));

        println!("✅ CollectionReference.document() works!");
    });
}

/// Test: Nested document paths
#[test]
fn test_nested_documents() {
    SHARED_RUNTIME.block_on(async {
        let firestore = get_firestore().await;
        let parent_path = format!("integration_tests/parent_{}", rand::random::<u32>());
        let child_path = format!("{}/subcollection/child", parent_path);

        // Create parent document
        let parent_ref = firestore.document(&parent_path);
        let parent_data = create_map(vec![("type", ValueType::StringValue("parent".to_string()))]);
        parent_ref
            .set(parent_data)
            .await
            .expect("Failed to create parent");

        // Create child document in subcollection
        let child_ref = firestore.document(&child_path);
        let child_data = create_map(vec![("type", ValueType::StringValue("child".to_string()))]);
        child_ref
            .set(child_data)
            .await
            .expect("Failed to create child");

        // Read child document
        let snapshot = child_ref.get().await.expect("Failed to read child");

        assert!(snapshot.exists(), "Child document should exist");

        // Verify child type field
        let child_type = snapshot.get("type").expect("type field missing");
        match &child_type.value_type {
            Some(ValueType::StringValue(s)) => {
                assert_eq!(s, "child", "Child type should be 'child'")
            }
            _ => panic!("Expected string value for type"),
        }

        // Clean up (delete child and parent)
        child_ref.delete().await.ok();
        parent_ref.delete().await.ok();

        println!("✅ Nested document paths work!");
    });
}

/// Test: Compound filters with And/Or logic
#[test]
fn test_compound_filters() {
    SHARED_RUNTIME.block_on(async {
        let firestore = get_firestore().await;
        let collection_path = format!("integration_tests_compound_{}", rand::random::<u32>());

        // Create test documents
        let test_docs = vec![
            ("doc1", 15, "inactive"),
            ("doc2", 25, "active"),
            ("doc3", 35, "active"),
            ("doc4", 45, "inactive"),
        ];

        for (doc_id, age, status) in &test_docs {
            let doc_path = format!("{}/{}", collection_path, doc_id);
            let doc_ref = firestore.document(&doc_path);
            let data = create_map(vec![
                ("age", ValueType::IntegerValue(*age)),
                ("status", ValueType::StringValue(status.to_string())),
            ]);
            doc_ref.set(data).await.expect("Failed to create document");
        }

        // Test And filter: age > 20 AND status == "active"
        // This should match doc2 (25, active) and doc3 (35, active)
        let age_value = Value {
            value_type: Some(ValueType::IntegerValue(20)),
        };
        let status_value = Value {
            value_type: Some(ValueType::StringValue("active".to_string())),
        };

        let and_filter = FilterCondition::And(vec![
            FilterCondition::GreaterThan("age".to_string(), age_value),
            FilterCondition::Equal("status".to_string(), status_value),
        ]);

        // Note: Actual query execution would require implementing query() method on CollectionReference
        // For now, we validate the filter structure
        match &and_filter {
            FilterCondition::And(filters) => {
                assert_eq!(filters.len(), 2);
                println!("✅ And filter structure: {} sub-filters", filters.len());
            }
            _ => panic!("Expected And filter"),
        }

        // Test Or filter: age < 20 OR age > 40
        // This should match doc1 (15) and doc4 (45)
        let age_20 = Value {
            value_type: Some(ValueType::IntegerValue(20)),
        };
        let age_40 = Value {
            value_type: Some(ValueType::IntegerValue(40)),
        };

        let or_filter = FilterCondition::Or(vec![
            FilterCondition::LessThan("age".to_string(), age_20),
            FilterCondition::GreaterThan("age".to_string(), age_40),
        ]);

        match &or_filter {
            FilterCondition::Or(filters) => {
                assert_eq!(filters.len(), 2);
                println!("✅ Or filter structure: {} sub-filters", filters.len());
            }
            _ => panic!("Expected Or filter"),
        }

        // Clean up all documents
        for (doc_id, _, _) in &test_docs {
            let doc_path = format!("{}/{}", collection_path, doc_id);
            firestore.document(&doc_path).delete().await.ok();
        }

        println!("✅ Compound filters (And/Or) work!");
    });
}

/// Test: Real-time listener using gRPC streaming
#[test]
fn test_snapshot_listener() {
    SHARED_RUNTIME.block_on(async {
        dotenvy::dotenv().ok();

        // Get authenticated firestore instance
        let firestore = get_firestore().await;
        let doc_path = test_doc_path("listener");

        // Get auth token and project info for listener
        let project_id = env::var("FIREBASE_PROJECT_ID").expect("FIREBASE_PROJECT_ID required");
        let database_id =
            env::var("FIREBASE_DATABASE_ID").unwrap_or_else(|_| "default".to_string());
        let api_key =
            env::var("FIREBASE_API_KEY").expect("FIREBASE_API_KEY must be set in .env file");
        let email = env::var("TEST_USER_EMAIL").expect("TEST_USER_EMAIL must be set in .env file");
        let password =
            env::var("TEST_USER_PASSWORD").expect("TEST_USER_PASSWORD must be set in .env file");

        // Get fresh auth token for listener (same as get_firestore() does)
        let app = App::create(AppOptions {
            api_key: api_key.clone(),
            project_id: project_id.clone(),
            app_name: None,
        })
        .await
        .expect("Failed to create App");

        let auth = Auth::get_auth(&app)
            .await
            .expect("Failed to get Auth instance");

        auth.sign_in_with_email_and_password(&email, &password)
            .await
            .expect("Failed to sign in");

        let user = auth
            .current_user()
            .await
            .expect("No current user after sign in");
        let auth_token = user
            .get_id_token(false)
            .await
            .expect("Failed to get ID token");

        // Create initial document
        let doc_ref = firestore.document(&doc_path);
        let initial_data = create_map(vec![("value", ValueType::IntegerValue(0))]);
        doc_ref
            .set(initial_data)
            .await
            .expect("Failed to create document");

        // Set up listener stream
        let mut stream = listen_document(
            &firestore,
            auth_token,
            project_id,
            database_id,
            doc_path.clone(),
            ListenerOptions::default(),
        )
        .await
        .expect("Failed to start listener");

        // Track updates received
        let updates = Arc::new(Mutex::new(Vec::new()));
        let updates_clone = updates.clone();

        // Spawn task to consume the stream
        let stream_task = tokio::spawn(async move {
            let mut count = 0;
            while let Some(result) = stream.next().await {
                if let Ok(snapshot) = result {
                    if let Some(data) = &snapshot.data {
                        if let Some(value_field) = data.fields.get("value") {
                            if let Some(ValueType::IntegerValue(val)) = &value_field.value_type {
                                updates_clone.lock().await.push(*val);
                                println!("📡 Listener received value: {}", val);
                                count += 1;
                                // Stop after receiving 3 updates (initial + 2 updates)
                                if count >= 3 {
                                    break;
                                }
                            }
                        }
                    }
                }
            }
        });

        // Wait a bit for initial snapshot
        tokio::time::sleep(tokio::time::Duration::from_millis(500)).await;

        // Update the document (should trigger listener)
        let update_data = create_map(vec![("value", ValueType::IntegerValue(42))]);
        doc_ref
            .set(update_data)
            .await
            .expect("Failed to update document");

        // Another update
        let update_data2 = create_map(vec![("value", ValueType::IntegerValue(100))]);
        doc_ref
            .set(update_data2)
            .await
            .expect("Failed to update document");

        // Wait for task to complete with timeout
        tokio::time::timeout(tokio::time::Duration::from_secs(5), stream_task)
            .await
            .expect("Timeout waiting for listener updates")
            .expect("Stream task panicked");

        // Verify we received updates
        let collected = updates.lock().await;
        println!("📊 Received {} updates: {:?}", collected.len(), *collected);

        assert!(
            !collected.is_empty(),
            "Should have received at least one update"
        );
        assert_eq!(
            collected.len(),
            3,
            "Should have received exactly 3 updates (initial + 2 changes)"
        );

        // Verify we got initial value (0), then update (42), then final update (100)
        assert_eq!(collected[0], 0, "First update should be initial value 0");
        assert_eq!(collected[1], 42, "Second update should be 42");
        assert_eq!(collected[2], 100, "Third update should be 100");

        // Clean up
        doc_ref.delete().await.expect("Failed to delete document");

        println!(
            "✅ Snapshot listener works! Received {} updates",
            collected.len()
        );
    });
}

/// Test: Get non-existent document returns NotFound
#[test]
fn test_get_nonexistent_document() {
    SHARED_RUNTIME.block_on(async {
        let firestore = get_firestore().await;
        let doc_path = test_doc_path("nonexistent");

        let doc_ref = firestore.document(&doc_path);
        let result = doc_ref.get().await;

        // Should either return error or snapshot with exists() == false
        match result {
            Err(e) => {
                println!("✅ Non-existent document returns error: {}", e);
            }
            Ok(snapshot) => {
                assert!(!snapshot.exists(), "Non-existent document should not exist");
                println!("✅ Non-existent document returns empty snapshot");
            }
        }
    });
}

/// Test: Update non-existent document should fail
#[test]
fn test_update_nonexistent_document() {
    SHARED_RUNTIME.block_on(async {
        let firestore = get_firestore().await;
        let doc_path = test_doc_path("update_nonexistent");

        let doc_ref = firestore.document(&doc_path);
        let update_data = create_map(vec![("field", ValueType::StringValue("value".to_string()))]);

        let result = doc_ref.update(update_data).await;

        // Update should fail if document doesn't exist
        assert!(
            result.is_err(),
            "Update should fail for non-existent document"
        );

        println!("✅ Update non-existent document fails as expected");
    });
}

/// Test: Delete non-existent document should succeed (idempotent)
#[test]
fn test_delete_nonexistent_document() {
    SHARED_RUNTIME.block_on(async {
        let firestore = get_firestore().await;
        let doc_path = test_doc_path("delete_nonexistent");

        let doc_ref = firestore.document(&doc_path);

        // Delete should succeed even if document doesn't exist (idempotent)
        doc_ref.delete().await.expect("Delete should be idempotent");

        println!("✅ Delete non-existent document is idempotent");
    });
}

/// Test: Document with single field (Firestore requires at least one field)
#[test]
fn test_minimal_document() {
    SHARED_RUNTIME.block_on(async {
        let firestore = get_firestore().await;
        let doc_path = test_doc_path("minimal");

        let doc_ref = firestore.document(&doc_path);

        // Firestore requires at least one field, so use a minimal document
        let minimal_data = create_map(vec![("exists", ValueType::BooleanValue(true))]);

        doc_ref
            .set(minimal_data)
            .await
            .expect("Failed to set minimal document");

        let snapshot = doc_ref.get().await.expect("Failed to get document");

        assert!(snapshot.exists(), "Document should exist");
        assert!(snapshot.data.is_some(), "Document should have data");
        assert_eq!(
            snapshot.data.as_ref().unwrap().fields.len(),
            1,
            "Document should have exactly 1 field"
        );

        let exists_field = snapshot.get("exists").expect("exists field missing");
        match &exists_field.value_type {
            Some(ValueType::BooleanValue(v)) => assert_eq!(*v, true, "exists field should be true"),
            _ => panic!("Expected boolean value for exists"),
        }

        // Clean up
        doc_ref.delete().await.expect("Failed to delete");

        println!("✅ Minimal document works!");
    });
}

/// Test: Document with various data types
#[test]
fn test_multiple_data_types() {
    SHARED_RUNTIME.block_on(async {
        let firestore = get_firestore().await;
        let doc_path = test_doc_path("data_types");

        let doc_ref = firestore.document(&doc_path);

        // Create document with various field types
        let data = create_map(vec![
            ("string", ValueType::StringValue("hello".to_string())),
            ("integer", ValueType::IntegerValue(42)),
            ("double", ValueType::DoubleValue(3.14)),
            ("boolean", ValueType::BooleanValue(true)),
            ("null", ValueType::NullValue(0)),
        ]);

        doc_ref
            .set(data)
            .await
            .expect("Failed to set document with multiple types");

        let snapshot = doc_ref.get().await.expect("Failed to get document");

        assert!(snapshot.exists(), "Document should exist");

        // Verify each type and value
        let string_val = snapshot.get("string").expect("string missing");
        match &string_val.value_type {
            Some(ValueType::StringValue(s)) => assert_eq!(s, "hello", "String should be 'hello'"),
            _ => panic!("Expected string value"),
        }

        let int_val = snapshot.get("integer").expect("integer missing");
        match &int_val.value_type {
            Some(ValueType::IntegerValue(i)) => assert_eq!(*i, 42, "Integer should be 42"),
            _ => panic!("Expected integer value"),
        }

        let double_val = snapshot.get("double").expect("double missing");
        match &double_val.value_type {
            Some(ValueType::DoubleValue(d)) => assert!(
                (d - 3.14).abs() < 0.001,
                "Double should be approximately 3.14"
            ),
            _ => panic!("Expected double value"),
        }

        let bool_val = snapshot.get("boolean").expect("boolean missing");
        match &bool_val.value_type {
            Some(ValueType::BooleanValue(b)) => assert_eq!(*b, true, "Boolean should be true"),
            _ => panic!("Expected boolean value"),
        }

        let null_val = snapshot.get("null").expect("null missing");
        match &null_val.value_type {
            Some(ValueType::NullValue(n)) => assert_eq!(*n, 0, "Null should be 0"),
            _ => panic!("Expected null value"),
        }

        // Clean up
        doc_ref.delete().await.expect("Failed to delete");

        println!("✅ Multiple data types work!");
    });
}

/// Test: Large document with many fields
#[test]
fn test_large_document() {
    SHARED_RUNTIME.block_on(async {
        let firestore = get_firestore().await;
        let doc_path = test_doc_path("large");

        let doc_ref = firestore.document(&doc_path);

        // Create document with 50 fields
        let mut fields = Vec::new();
        for i in 0..50 {
            fields.push((
                format!("field_{}", i).leak() as &str,
                ValueType::IntegerValue(i),
            ));
        }

        let data = create_map(fields);
        doc_ref
            .set(data)
            .await
            .expect("Failed to set large document");

        let snapshot = doc_ref.get().await.expect("Failed to get large document");

        assert!(snapshot.exists(), "Document should exist");
        assert_eq!(
            snapshot.data.as_ref().unwrap().fields.len(),
            50,
            "Document should have 50 fields"
        );

        // Verify first, middle, and last fields
        let field_0 = snapshot.get("field_0").expect("field_0 missing");
        match &field_0.value_type {
            Some(ValueType::IntegerValue(v)) => assert_eq!(*v, 0, "field_0 should be 0"),
            _ => panic!("Expected integer value for field_0"),
        }

        let field_25 = snapshot.get("field_25").expect("field_25 missing");
        match &field_25.value_type {
            Some(ValueType::IntegerValue(v)) => assert_eq!(*v, 25, "field_25 should be 25"),
            _ => panic!("Expected integer value for field_25"),
        }

        let field_49 = snapshot.get("field_49").expect("field_49 missing");
        match &field_49.value_type {
            Some(ValueType::IntegerValue(v)) => assert_eq!(*v, 49, "field_49 should be 49"),
            _ => panic!("Expected integer value for field_49"),
        }

        // Clean up
        doc_ref.delete().await.expect("Failed to delete");

        println!("✅ Large document (50 fields) works!");
    });
}

/// Test: Overwrite document with set()
#[test]
fn test_overwrite_document() {
    SHARED_RUNTIME.block_on(async {
        let firestore = get_firestore().await;
        let doc_path = test_doc_path("overwrite");

        let doc_ref = firestore.document(&doc_path);

        // Create initial document
        let initial = create_map(vec![
            ("field1", ValueType::StringValue("value1".to_string())),
            ("field2", ValueType::IntegerValue(100)),
        ]);
        doc_ref.set(initial).await.expect("Failed to set initial");

        // Overwrite with completely new data (set replaces entire document)
        let new_data = create_map(vec![("field3", ValueType::BooleanValue(true))]);
        doc_ref.set(new_data).await.expect("Failed to overwrite");

        let snapshot = doc_ref.get().await.expect("Failed to get");

        // Old fields should be gone
        assert!(snapshot.get("field1").is_none());
        assert!(snapshot.get("field2").is_none());

        // New field should exist
        assert!(snapshot.get("field3").is_some());

        // Clean up
        doc_ref.delete().await.expect("Failed to delete");

        println!("✅ Document overwrite works!");
    });
}

/// Test: Batch with mixed operations
#[test]
fn test_batch_mixed_operations() {
    SHARED_RUNTIME.block_on(async {
        let firestore = get_firestore().await;
        let collection = format!("integration_tests_batch_mixed_{}", rand::random::<u32>());

        // Create one document first
        let doc1_path = format!("{}/doc1", collection);
        let doc1 = firestore.document(&doc1_path);
        doc1.set(create_map(vec![("value", ValueType::IntegerValue(1))]))
            .await
            .expect("Failed to create doc1");

        // Create batch with mixed operations
        let batch = firestore.batch();

        // Set doc2 (create new)
        let batch = batch.set(
            format!("{}/doc2", collection),
            create_map(vec![("value", ValueType::IntegerValue(2))]),
        );

        // Update doc1 (modify existing)
        let batch = batch.update(
            doc1_path.clone(),
            create_map(vec![("value", ValueType::IntegerValue(10))]),
        );

        // Set doc3 (create new)
        let batch = batch.set(
            format!("{}/doc3", collection),
            create_map(vec![("value", ValueType::IntegerValue(3))]),
        );

        // Delete doc2 (delete what we just created in this batch)
        let batch = batch.delete(format!("{}/doc2", collection));

        // Commit batch
        batch.commit().await.expect("Failed to commit mixed batch");

        // Verify results
        let doc1_snapshot = doc1.get().await.expect("Failed to get doc1");
        let value1 = doc1_snapshot
            .get("value")
            .expect("value field missing in doc1");
        match &value1.value_type {
            Some(ValueType::IntegerValue(v)) => {
                assert_eq!(*v, 10, "doc1 value should be updated to 10")
            }
            _ => panic!("Expected integer value for doc1"),
        }

        let doc2 = firestore.document(&format!("{}/doc2", collection));
        let doc2_result = doc2.get().await;
        assert!(
            doc2_result.is_err() || !doc2_result.unwrap().exists(),
            "doc2 should be deleted"
        );

        let doc3 = firestore.document(&format!("{}/doc3", collection));
        let doc3_snapshot = doc3.get().await.expect("Failed to get doc3");
        assert!(doc3_snapshot.exists(), "doc3 should exist");
        let value3 = doc3_snapshot
            .get("value")
            .expect("value field missing in doc3");
        match &value3.value_type {
            Some(ValueType::IntegerValue(v)) => assert_eq!(*v, 3, "doc3 value should be 3"),
            _ => panic!("Expected integer value for doc3"),
        }

        // Clean up
        doc1.delete().await.ok();
        doc3.delete().await.ok();

        println!("✅ Batch with mixed operations works!");
    });
}

/// Test: Empty batch should fail with InvalidArgument
#[test]
fn test_empty_batch() {
    SHARED_RUNTIME.block_on(async {
        let firestore = get_firestore().await;

        let batch = firestore.batch();

        // Firestore rejects empty batches
        let result = batch.commit().await;
        assert!(result.is_err(), "Empty batch should fail");

        match result {
            Err(e) => {
                let err_str = format!("{:?}", e);
                assert!(
                    err_str.contains("InvalidArgument") || err_str.contains("empty"),
                    "Error should mention empty batch: {}",
                    err_str
                );
                println!("✅ Empty batch fails as expected: {}", e);
            }
            Ok(_) => panic!("Empty batch should not succeed"),
        }
    });
}

/// Test: Collection path parsing and validation
#[test]
fn test_collection_paths() {
    SHARED_RUNTIME.block_on(async {
        let firestore = get_firestore().await;

        // Simple collection
        let col1 = firestore.collection("users");
        let doc1 = col1.document("alice");
        assert_eq!(doc1.path, "users/alice");
        assert_eq!(doc1.id(), "alice");

        // Nested collection (subcollection) - use full path
        let post = firestore.document("users/bob/posts/post1");
        assert_eq!(post.path, "users/bob/posts/post1");
        assert_eq!(post.id(), "post1");
        assert_eq!(post.parent_path(), Some("users/bob/posts"));

        println!("✅ Collection path parsing works!");
    });
}

/// Test: Document ID extraction
#[test]
fn test_document_id_extraction() {
    SHARED_RUNTIME.block_on(async {
        let firestore = get_firestore().await;

        let doc1 = firestore.document("users/alice");
        assert_eq!(doc1.id(), "alice");

        let doc2 = firestore.document("projects/proj1/tasks/task2");
        assert_eq!(doc2.id(), "task2");

        let doc3 = firestore.document("single");
        assert_eq!(doc3.id(), "single");

        println!("✅ Document ID extraction works!");
    });
}

/// Test: Special characters in document IDs
#[test]
fn test_special_characters_in_paths() {
    SHARED_RUNTIME.block_on(async {
        let firestore = get_firestore().await;

        // Test with underscores, hyphens, periods
        let doc_id = "test_doc-123.v2";
        let doc_path = format!("integration_tests/{}", doc_id);
        let doc_ref = firestore.document(&doc_path);

        let data = create_map(vec![("test", ValueType::BooleanValue(true))]);

        doc_ref
            .set(data)
            .await
            .expect("Failed to set document with special chars");

        let snapshot = doc_ref
            .get()
            .await
            .expect("Failed to get document with special chars");

        assert!(snapshot.exists());
        assert_eq!(snapshot.id(), doc_id);

        // Clean up
        doc_ref.delete().await.expect("Failed to delete");

        println!("✅ Special characters in paths work!");
    });
}

/// Test: Listener receives delete events
#[test]
fn test_listener_delete_event() {
    SHARED_RUNTIME.block_on(async {
        dotenvy::dotenv().ok();

        let firestore = get_firestore().await;
        let doc_path = test_doc_path("listener_delete");

        // Get credentials for listener
        let project_id = env::var("FIREBASE_PROJECT_ID").expect("FIREBASE_PROJECT_ID required");
        let database_id =
            env::var("FIREBASE_DATABASE_ID").unwrap_or_else(|_| "default".to_string());
        let api_key = env::var("FIREBASE_API_KEY").expect("FIREBASE_API_KEY required");
        let email = env::var("TEST_USER_EMAIL").expect("TEST_USER_EMAIL required");
        let password = env::var("TEST_USER_PASSWORD").expect("TEST_USER_PASSWORD required");

        let app = App::create(AppOptions {
            api_key,
            project_id: project_id.clone(),
            app_name: None,
        })
        .await
        .expect("Failed to create App");

        let auth = Auth::get_auth(&app).await.expect("Failed to get Auth");
        auth.sign_in_with_email_and_password(&email, &password)
            .await
            .expect("Failed to sign in");

        let user = auth.current_user().await.expect("No user");
        let auth_token = user.get_id_token(false).await.expect("Failed to get token");

        // Create initial document
        let doc_ref = firestore.document(&doc_path);
        doc_ref
            .set(create_map(vec![(
                "status",
                ValueType::StringValue("active".to_string()),
            )]))
            .await
            .expect("Failed to create");

        // Start listener
        let mut stream = listen_document(
            &firestore,
            auth_token,
            project_id,
            database_id,
            doc_path.clone(),
            ListenerOptions::default(),
        )
        .await
        .expect("Failed to start listener");

        let received_delete = Arc::new(Mutex::new(false));
        let delete_flag = received_delete.clone();

        let stream_task = tokio::spawn(async move {
            let mut got_initial = false;
            while let Some(result) = stream.next().await {
                if let Ok(snapshot) = result {
                    if !got_initial {
                        got_initial = true;
                        // Skip initial snapshot
                        continue;
                    }
                    if !snapshot.exists() {
                        *delete_flag.lock().await = true;
                        println!("📡 Listener received delete event");
                        break;
                    }
                }
            }
        });

        // Wait briefly for initial snapshot
        tokio::time::sleep(tokio::time::Duration::from_millis(500)).await;

        // Delete the document
        doc_ref.delete().await.expect("Failed to delete");

        // Wait for task to complete with timeout
        tokio::time::timeout(tokio::time::Duration::from_secs(3), stream_task)
            .await
            .expect("Timeout waiting for delete event")
            .expect("Stream task panicked");

        let got_delete = *received_delete.lock().await;
        assert!(got_delete, "Listener should receive delete event");

        println!("✅ Listener delete event works!");
    });
}

/// Test: Concurrent writes to same document
#[test]
fn test_concurrent_writes() {
    SHARED_RUNTIME.block_on(async {
        let firestore = get_firestore().await;
        let doc_path = test_doc_path("concurrent");

        let doc_ref = firestore.document(&doc_path);

        // Create initial document
        doc_ref
            .set(create_map(vec![("counter", ValueType::IntegerValue(0))]))
            .await
            .expect("Failed to create");

        // Spawn multiple concurrent updates
        let mut handles = vec![];
        for i in 1..=5 {
            let doc_ref_clone = doc_ref.clone();
            let handle = tokio::spawn(async move {
                let data = create_map(vec![
                    ("counter", ValueType::IntegerValue(i)),
                    ("writer", ValueType::IntegerValue(i)),
                ]);
                doc_ref_clone.set(data).await
            });
            handles.push(handle);
        }

        // Wait for all writes to complete
        for handle in handles {
            handle.await.expect("Task panicked").expect("Write failed");
        }

        // Read final state (one of the writes should have won)
        let snapshot = doc_ref.get().await.expect("Failed to get");
        assert!(
            snapshot.exists(),
            "Document should exist after concurrent writes"
        );

        let counter = snapshot.get("counter").expect("counter missing");
        match &counter.value_type {
            Some(ValueType::IntegerValue(v)) => {
                assert!(
                    *v >= 1 && *v <= 5,
                    "Counter should be one of the written values (1-5), got {}",
                    v
                );
            }
            _ => panic!("Expected integer value for counter"),
        }

        // Verify writer field matches counter (both from same write)
        let writer = snapshot.get("writer").expect("writer missing");
        match (&counter.value_type, &writer.value_type) {
            (Some(ValueType::IntegerValue(c)), Some(ValueType::IntegerValue(w))) => {
                assert_eq!(
                    c, w,
                    "Counter and writer should match (from same write operation)"
                );
            }
            _ => panic!("Expected integer values for counter and writer"),
        }

        // Clean up
        doc_ref.delete().await.expect("Failed to delete");

        println!("✅ Concurrent writes complete (last write wins)!");
    });
}

/// Test: Very long document path
#[test]
fn test_deep_nested_path() {
    SHARED_RUNTIME.block_on(async {
        let firestore = get_firestore().await;

        // Create a very deep path (10 levels)
        let path = format!(
            "integration_tests/l1_{}/level2/l2_{}/level3/l3_{}/level4/l4_{}/level5/l5_{}",
            rand::random::<u32>(),
            rand::random::<u32>(),
            rand::random::<u32>(),
            rand::random::<u32>(),
            rand::random::<u32>()
        );

        let doc_ref = firestore.document(&path);
        doc_ref
            .set(create_map(vec![("depth", ValueType::IntegerValue(5))]))
            .await
            .expect("Failed to set deep document");

        let snapshot = doc_ref.get().await.expect("Failed to get deep document");
        assert!(snapshot.exists(), "Deep nested document should exist");

        // Verify depth field value
        let depth = snapshot.get("depth").expect("depth field missing");
        match &depth.value_type {
            Some(ValueType::IntegerValue(v)) => assert_eq!(*v, 5, "depth field should be 5"),
            _ => panic!("Expected integer value for depth"),
        }

        // Clean up
        doc_ref.delete().await.expect("Failed to delete");

        println!("✅ Deep nested paths work!");
    });
}

/// Test: Document listener receives initial snapshot and updates
#[test]
fn test_document_listener_receives_updates() {
    SHARED_RUNTIME.block_on(async {
        let firestore = get_firestore().await;
        let doc_path = test_doc_path("listen_doc");

        // Create initial document
        let doc_ref = firestore.document(&doc_path);
        doc_ref
            .set(create_map(vec![
                ("counter", ValueType::IntegerValue(0)),
                ("name", ValueType::StringValue("test".to_string())),
            ]))
            .await
            .expect("Failed to create document");

        // Start listening
        let mut stream = doc_ref.listen(None);

        // Should receive initial snapshot
        let snapshot = tokio::time::timeout(std::time::Duration::from_secs(10), stream.next())
            .await
            .expect("Timeout waiting for initial snapshot")
            .expect("Stream ended")
            .expect("Error in initial snapshot");

        assert!(snapshot.exists(), "Initial snapshot should exist");

        // Verify initial data - counter should be 0
        let counter = snapshot.get("counter").expect("counter field missing");
        match &counter.value_type {
            Some(ValueType::IntegerValue(v)) => assert_eq!(*v, 0, "Initial counter should be 0"),
            _ => panic!("Expected integer value for counter"),
        }

        // Verify initial name is "test"
        let name = snapshot.get("name").expect("name field missing");
        match &name.value_type {
            Some(ValueType::StringValue(s)) => {
                assert_eq!(s, "test", "Initial name should be 'test'")
            }
            _ => panic!("Expected string value for name"),
        }

        // Update document
        doc_ref
            .set(create_map(vec![
                ("counter", ValueType::IntegerValue(1)),
                ("name", ValueType::StringValue("updated".to_string())),
            ]))
            .await
            .expect("Failed to update document");

        // Should receive update
        let updated_snapshot =
            tokio::time::timeout(std::time::Duration::from_secs(10), stream.next())
                .await
                .expect("Timeout waiting for update")
                .expect("Stream ended")
                .expect("Error in update snapshot");

        assert!(updated_snapshot.exists(), "Updated snapshot should exist");

        // Verify updated counter is 1
        let counter = updated_snapshot
            .get("counter")
            .expect("counter field missing in update");
        match &counter.value_type {
            Some(ValueType::IntegerValue(v)) => assert_eq!(*v, 1, "Updated counter should be 1"),
            _ => panic!("Expected integer value for counter"),
        }

        // Verify updated name is "updated"
        let name = updated_snapshot
            .get("name")
            .expect("name field missing in update");
        match &name.value_type {
            Some(ValueType::StringValue(s)) => {
                assert_eq!(s, "updated", "Updated name should be 'updated'")
            }
            _ => panic!("Expected string value for name"),
        }

        // Clean up
        drop(stream);
        doc_ref.delete().await.expect("Failed to delete");

        println!("✅ Document listener receives updates!");
    });
}

/// Test: Listener receives delete event
#[test]
fn test_document_listener_receives_delete() {
    SHARED_RUNTIME.block_on(async {
        let firestore = get_firestore().await;
        let doc_path = test_doc_path("listen_delete");

        // Create document
        let doc_ref = firestore.document(&doc_path);
        doc_ref
            .set(create_map(vec![("temp", ValueType::BooleanValue(true))]))
            .await
            .expect("Failed to create document");

        // Start listening
        let mut stream = doc_ref.listen(None);

        // Receive initial snapshot
        let snapshot = tokio::time::timeout(std::time::Duration::from_secs(10), stream.next())
            .await
            .expect("Timeout waiting for initial snapshot")
            .expect("Stream ended")
            .expect("Error in snapshot");

        assert!(snapshot.exists(), "Initial snapshot should exist");

        // Verify initial field value
        let temp = snapshot.get("temp").expect("temp field missing");
        match &temp.value_type {
            Some(ValueType::BooleanValue(v)) => assert_eq!(*v, true, "temp field should be true"),
            _ => panic!("Expected boolean value for temp"),
        };

        // Delete document
        doc_ref.delete().await.expect("Failed to delete");

        // Should receive delete event (snapshot with exists=false)
        let deleted_snapshot =
            tokio::time::timeout(std::time::Duration::from_secs(10), stream.next())
                .await
                .expect("Timeout waiting for delete")
                .expect("Stream ended")
                .expect("Error in delete snapshot");

        assert!(
            !deleted_snapshot.exists(),
            "Snapshot should not exist after delete"
        );

        drop(stream);

        println!("✅ Document listener receives delete event!");
    });
}

/// Test: Multiple listeners on same document
#[test]
fn test_multiple_document_listeners() {
    SHARED_RUNTIME.block_on(async {
        let firestore = get_firestore().await;
        let doc_path = test_doc_path("listen_multi");

        // Create document
        let doc_ref = firestore.document(&doc_path);
        doc_ref
            .set(create_map(vec![("value", ValueType::IntegerValue(100))]))
            .await
            .expect("Failed to create document");

        // Start two listeners
        let mut stream1 = doc_ref.listen(None);
        let mut stream2 = doc_ref.listen(None);

        // Both should receive initial snapshot
        let snap1 = tokio::time::timeout(std::time::Duration::from_secs(10), stream1.next())
            .await
            .expect("Timeout on stream1")
            .expect("Stream1 ended")
            .expect("Error on stream1");

        let snap2 = tokio::time::timeout(std::time::Duration::from_secs(10), stream2.next())
            .await
            .expect("Timeout on stream2")
            .expect("Stream2 ended")
            .expect("Error on stream2");

        assert!(snap1.exists(), "Initial snapshot 1 should exist");
        assert!(snap2.exists(), "Initial snapshot 2 should exist");

        // Verify both listeners received initial value of 100
        let val1_init = snap1.get("value").expect("value missing in snap1");
        match &val1_init.value_type {
            Some(ValueType::IntegerValue(v)) => {
                assert_eq!(*v, 100, "Listener 1 initial value should be 100")
            }
            _ => panic!("Expected integer value"),
        }

        let val2_init = snap2.get("value").expect("value missing in snap2");
        match &val2_init.value_type {
            Some(ValueType::IntegerValue(v)) => {
                assert_eq!(*v, 100, "Listener 2 initial value should be 100")
            }
            _ => panic!("Expected integer value"),
        }

        // Update document
        doc_ref
            .set(create_map(vec![("value", ValueType::IntegerValue(200))]))
            .await
            .expect("Failed to update");

        // Both should receive update
        let update1 = tokio::time::timeout(std::time::Duration::from_secs(10), stream1.next())
            .await
            .expect("Timeout on stream1 update")
            .expect("Stream1 ended")
            .expect("Error on stream1 update");

        let update2 = tokio::time::timeout(std::time::Duration::from_secs(10), stream2.next())
            .await
            .expect("Timeout on stream2 update")
            .expect("Stream2 ended")
            .expect("Error on stream2 update");

        // Verify both listeners got the updated value of 200
        let val1 = update1.get("value").expect("value missing in update1");
        let val2 = update2.get("value").expect("value missing in update2");

        match (&val1.value_type, &val2.value_type) {
            (Some(ValueType::IntegerValue(v1)), Some(ValueType::IntegerValue(v2))) => {
                assert_eq!(*v1, 200, "Listener 1 should receive updated value 200");
                assert_eq!(*v2, 200, "Listener 2 should receive updated value 200");
            }
            _ => panic!("Expected integer values"),
        }

        // Clean up
        drop(stream1);
        drop(stream2);
        doc_ref.delete().await.expect("Failed to delete");

        println!("✅ Multiple document listeners work!");
    });
}

/// Test: Listener stops receiving updates after drop
#[test]
fn test_listener_cleanup_on_drop() {
    SHARED_RUNTIME.block_on(async {
        let firestore = get_firestore().await;
        let doc_path = test_doc_path("listen_cleanup");

        // Create document
        let doc_ref = firestore.document(&doc_path);
        doc_ref
            .set(create_map(vec![("count", ValueType::IntegerValue(0))]))
            .await
            .expect("Failed to create document");

        {
            let mut stream = doc_ref.listen(None);

            // Receive initial snapshot
            let snapshot = tokio::time::timeout(std::time::Duration::from_secs(10), stream.next())
                .await
                .expect("Timeout")
                .expect("Stream ended")
                .expect("Error");

            // Verify initial count is 0
            assert!(snapshot.exists(), "Initial snapshot should exist");
            let count = snapshot.get("count").expect("count field missing");
            match &count.value_type {
                Some(ValueType::IntegerValue(v)) => assert_eq!(*v, 0, "Initial count should be 0"),
                _ => panic!("Expected integer value for count"),
            }

            // Stream dropped here
        }

        // Update document after listener dropped
        doc_ref
            .set(create_map(vec![("count", ValueType::IntegerValue(1))]))
            .await
            .expect("Failed to update");

        // Small delay to ensure no events are being processed
        tokio::time::sleep(std::time::Duration::from_millis(500)).await;

        // Clean up
        doc_ref.delete().await.expect("Failed to delete");

        println!("✅ Listener cleanup on drop works!");
    });
}

#[test]
fn test_query_listener_receives_updates() {
    SHARED_RUNTIME.block_on(async {
        use firebase_rust_sdk::firestore::Query;
        use futures::stream::StreamExt;

        let firestore = get_firestore().await;
        let test_id = std::time::SystemTime::now()
            .duration_since(std::time::UNIX_EPOCH)
            .unwrap()
            .as_millis();
        let collection_name = format!("query_listener_updates_{}", test_id);
        let collection = firestore.collection(&collection_name);

        // Create initial document
        let doc1 = firestore.document(&format!("{}/doc1", collection_name));
        doc1.set(create_map(vec![
            ("name", ValueType::StringValue("Alice".to_string())),
            ("count", ValueType::IntegerValue(0)),
        ]))
        .await
        .expect("Failed to create doc1");

        // Start query listener
        let mut stream = collection.listen(None);

        // Receive initial snapshot
        let snapshot = tokio::time::timeout(std::time::Duration::from_secs(10), stream.next())
            .await
            .expect("Timeout waiting for initial snapshot")
            .expect("Stream ended")
            .expect("Error");

        assert_eq!(
            snapshot.len(),
            1,
            "Initial snapshot should contain 1 document"
        );
        let docs = snapshot.documents();
        let initial_count = docs[0].get("count").expect("count field missing");
        match &initial_count.value_type {
            Some(ValueType::IntegerValue(v)) => assert_eq!(*v, 0, "Initial count should be 0"),
            _ => panic!("Expected integer value for count"),
        }

        // Update the document - this should trigger a listener update
        doc1.set(create_map(vec![
            ("name", ValueType::StringValue("Alice".to_string())),
            ("count", ValueType::IntegerValue(1)),
        ]))
        .await
        .expect("Failed to update doc1");

        // Receive update snapshot
        let snapshot = tokio::time::timeout(std::time::Duration::from_secs(10), stream.next())
            .await
            .expect("Timeout waiting for update")
            .expect("Stream ended")
            .expect("Error");

        assert_eq!(
            snapshot.len(),
            1,
            "Updated snapshot should contain 1 document"
        );
        let docs = snapshot.documents();
        let updated_count = docs[0].get("count").expect("count field missing");
        match &updated_count.value_type {
            Some(ValueType::IntegerValue(v)) => assert_eq!(*v, 1, "Count should be updated to 1"),
            _ => panic!("Expected integer value for count"),
        }

        // Add a new document - this should trigger another update
        let doc2 = firestore.document(&format!("{}/doc2", collection_name));
        doc2.set(create_map(vec![
            ("name", ValueType::StringValue("Bob".to_string())),
            ("count", ValueType::IntegerValue(5)),
        ]))
        .await
        .expect("Failed to create doc2");

        // Receive snapshot with new document
        let snapshot = tokio::time::timeout(std::time::Duration::from_secs(10), stream.next())
            .await
            .expect("Timeout waiting for new document")
            .expect("Stream ended")
            .expect("Error");

        assert_eq!(snapshot.len(), 2, "Snapshot should now contain 2 documents");

        let names: Vec<String> = snapshot
            .documents()
            .iter()
            .map(|doc| {
                let name_value = doc.get("name").expect("name field missing");
                match &name_value.value_type {
                    Some(ValueType::StringValue(s)) => s.clone(),
                    _ => panic!("Expected string value for name"),
                }
            })
            .collect();

        assert!(names.contains(&"Alice".to_string()), "Should contain Alice");
        assert!(names.contains(&"Bob".to_string()), "Should contain Bob");

        // Clean up
        doc1.delete().await.expect("Failed to delete doc1");
        doc2.delete().await.expect("Failed to delete doc2");

        println!("✅ Query listener receives real-time updates!");
    });
}

#[test]
fn test_query_listener_with_filter_updates() {
    SHARED_RUNTIME.block_on(async {
        use firebase_rust_sdk::firestore::Query;
        use futures::stream::StreamExt;

        let firestore = get_firestore().await;
        let test_id = std::time::SystemTime::now()
            .duration_since(std::time::UNIX_EPOCH)
            .unwrap()
            .as_millis();
        let collection_name = format!("query_listener_filter_{}", test_id);
        let collection = firestore.collection(&collection_name);

        // Create test documents - only doc2 matches filter initially (age > 25)
        let doc1 = firestore.document(&format!("{}/doc1", collection_name));
        doc1.set(create_map(vec![
            ("name", ValueType::StringValue("Alice".to_string())),
            ("age", ValueType::IntegerValue(25)),
        ]))
        .await
        .expect("Failed to create doc1");

        let doc2 = firestore.document(&format!("{}/doc2", collection_name));
        doc2.set(create_map(vec![
            ("name", ValueType::StringValue("Bob".to_string())),
            ("age", ValueType::IntegerValue(30)),
        ]))
        .await
        .expect("Failed to create doc2");

        // Start filtered query listener (age > 25)
        let age_value = Value {
            value_type: Some(ValueType::IntegerValue(25)),
        };
        let query = collection.where_greater_than("age", age_value);
        let mut stream = query.listen(None);

        // Receive initial snapshot - should only have Bob
        let snapshot = tokio::time::timeout(std::time::Duration::from_secs(10), stream.next())
            .await
            .expect("Timeout waiting for initial snapshot")
            .expect("Stream ended")
            .expect("Error");

        assert_eq!(
            snapshot.len(),
            1,
            "Should contain 1 document matching filter (Bob)"
        );
        let names: Vec<String> = snapshot
            .documents()
            .iter()
            .map(|doc| {
                let name_value = doc.get("name").expect("name field missing");
                match &name_value.value_type {
                    Some(ValueType::StringValue(s)) => s.clone(),
                    _ => panic!("Expected string value for name"),
                }
            })
            .collect();
        assert!(
            names.contains(&"Bob".to_string()),
            "Initial should contain Bob"
        );

        // Add doc3 which matches the filter - should appear in query results
        let doc3 = firestore.document(&format!("{}/doc3", collection_name));
        doc3.set(create_map(vec![
            ("name", ValueType::StringValue("Charlie".to_string())),
            ("age", ValueType::IntegerValue(35)),
        ]))
        .await
        .expect("Failed to create doc3");

        // Receive update with Charlie added
        let snapshot = tokio::time::timeout(std::time::Duration::from_secs(10), stream.next())
            .await
            .expect("Timeout waiting for Charlie")
            .expect("Stream ended")
            .expect("Error");

        assert_eq!(
            snapshot.len(),
            2,
            "Should now contain 2 documents (Bob and Charlie)"
        );
        let names: Vec<String> = snapshot
            .documents()
            .iter()
            .map(|doc| {
                let name_value = doc.get("name").expect("name field missing");
                match &name_value.value_type {
                    Some(ValueType::StringValue(s)) => s.clone(),
                    _ => panic!("Expected string value for name"),
                }
            })
            .collect();
        assert!(names.contains(&"Bob".to_string()), "Should contain Bob");
        assert!(
            names.contains(&"Charlie".to_string()),
            "Should contain Charlie"
        );

        // Update Alice to match filter (age 25 -> 40) - should now appear
        doc1.set(create_map(vec![
            ("name", ValueType::StringValue("Alice".to_string())),
            ("age", ValueType::IntegerValue(40)),
        ]))
        .await
        .expect("Failed to update doc1");

        // Receive update with Alice now matching filter
        let snapshot = tokio::time::timeout(std::time::Duration::from_secs(10), stream.next())
            .await
            .expect("Timeout waiting for Alice update")
            .expect("Stream ended")
            .expect("Error");

        assert_eq!(
            snapshot.len(),
            3,
            "Should now contain 3 documents (Alice, Bob, Charlie)"
        );
        let names: Vec<String> = snapshot
            .documents()
            .iter()
            .map(|doc| {
                let name_value = doc.get("name").expect("name field missing");
                match &name_value.value_type {
                    Some(ValueType::StringValue(s)) => s.clone(),
                    _ => panic!("Expected string value for name"),
                }
            })
            .collect();
        assert!(
            names.contains(&"Alice".to_string()),
            "Should now contain Alice (age updated to 40)"
        );
        assert!(names.contains(&"Bob".to_string()), "Should contain Bob");
        assert!(
            names.contains(&"Charlie".to_string()),
            "Should contain Charlie"
        );

        // Clean up
        doc1.delete().await.expect("Failed to delete doc1");
        doc2.delete().await.expect("Failed to delete doc2");
        doc3.delete().await.expect("Failed to delete doc3");

        println!("✅ Query listener with filter receives real-time updates!");
    });
}

#[test]
fn test_query_listener_document_removal() {
    SHARED_RUNTIME.block_on(async {
        use firebase_rust_sdk::firestore::Query;
        use futures::stream::StreamExt;

        let firestore = get_firestore().await;
        let test_id = std::time::SystemTime::now()
            .duration_since(std::time::UNIX_EPOCH)
            .unwrap()
            .as_millis();
        let collection_name = format!("query_listener_removal_{}", test_id);
        let collection = firestore.collection(&collection_name);

        // Create two documents
        let doc1 = firestore.document(&format!("{}/doc1", collection_name));
        doc1.set(create_map(vec![(
            "name",
            ValueType::StringValue("Alice".to_string()),
        )]))
        .await
        .expect("Failed to create doc1");

        let doc2 = firestore.document(&format!("{}/doc2", collection_name));
        doc2.set(create_map(vec![(
            "name",
            ValueType::StringValue("Bob".to_string()),
        )]))
        .await
        .expect("Failed to create doc2");

        // Start query listener
        let mut stream = collection.listen(None);

        // Receive initial snapshot with both documents
        let snapshot = tokio::time::timeout(std::time::Duration::from_secs(10), stream.next())
            .await
            .expect("Timeout waiting for initial snapshot")
            .expect("Stream ended")
            .expect("Error");

        assert_eq!(
            snapshot.len(),
            2,
            "Initial snapshot should contain 2 documents"
        );

        // Delete one document - should trigger update
        doc1.delete().await.expect("Failed to delete doc1");

        // Receive update with document removed
        let snapshot = tokio::time::timeout(std::time::Duration::from_secs(15), stream.next())
            .await
            .expect("Timeout waiting for removal update")
            .expect("Stream ended")
            .expect("Error");

        assert_eq!(snapshot.len(), 1, "Should now contain only 1 document");
        let docs = snapshot.documents();
        let name_value = docs[0].get("name").expect("name field missing");
        match &name_value.value_type {
            Some(ValueType::StringValue(s)) => {
                assert_eq!(s, "Bob", "Remaining document should be Bob")
            }
            _ => panic!("Expected string value for name"),
        }

        // Delete the last document
        doc2.delete().await.expect("Failed to delete doc2");

        // Receive update with empty results
        let snapshot = tokio::time::timeout(std::time::Duration::from_secs(15), stream.next())
            .await
            .expect("Timeout waiting for empty update")
            .expect("Stream ended")
            .expect("Error");

        assert_eq!(snapshot.len(), 0, "Should now be empty");
        assert!(snapshot.is_empty(), "Snapshot should be empty");

        println!("✅ Query listener detects document removals!");
    });
}

#[test]
fn test_count_aggregation() {
    SHARED_RUNTIME.block_on(async {
        let firestore = get_firestore().await;
        let unique_id = format!("count_test_{}", rand::thread_rng().gen::<u32>());
        let collection = firestore.collection(&unique_id);

        // Create test documents
        for i in 0..5 {
            let doc_id = format!("doc_{}", i);
            let data = create_map(vec![
                ("counter", ValueType::IntegerValue(i as i64)),
                ("category", ValueType::StringValue("A".to_string())),
            ]);

            collection
                .document(&doc_id)
                .set(data)
                .await
                .expect("Failed to set document");
        }

        // Test count aggregation  - CollectionReference implements Query trait
        use firebase_rust_sdk::firestore::Query;
        let aggregate_query = collection.count();

        let snapshot = aggregate_query
            .get()
            .await
            .expect("Failed to execute aggregate query");

        let count = snapshot
            .get("count")
            .expect("Count not found in results");

        match count.value_type {
            Some(ValueType::IntegerValue(n)) => {
                assert_eq!(n, 5, "Expected 5 documents");
            }
            _ => panic!("Count should be an integer, got {:?}", count),
        }

        // Cleanup
        for i in 0..5 {
            let doc_id = format!("doc_{}", i);
            collection
                .document(&doc_id)
                .delete()
                .await
                .expect("Failed to delete document");
        }

        println!("✅ Count aggregation works!");
    });
}

#[test]
fn test_sum_aggregation() {
    SHARED_RUNTIME.block_on(async {
        let firestore = get_firestore().await;
        let unique_id = format!("sum_test_{}", rand::thread_rng().gen::<u32>());
        let collection = firestore.collection(&unique_id);

        // Add test documents with numeric values
        for i in 1..=5 {
            let doc_id = format!("doc_{}", i);
            let data = create_map(vec![
                ("amount", ValueType::IntegerValue(i)),
            ]);

            collection
                .document(&doc_id)
                .set(data)
                .await
                .expect("Failed to set document");
        }

        // Test sum aggregation - CollectionReference implements Query trait
        use firebase_rust_sdk::firestore::{AggregateField, Query};
        let sum_field = AggregateField::sum("amount");
        let aggregate_query = collection.aggregate(vec![sum_field]);

        let snapshot = aggregate_query
            .get()
            .await
            .expect("Failed to execute aggregate query");

        let sum = snapshot
            .get("sum_amount")
            .expect("Sum not found in results");

        match sum.value_type {
            Some(ValueType::IntegerValue(n)) => {
                assert_eq!(n, 15, "Expected sum of 1+2+3+4+5 = 15");
            }
            Some(ValueType::DoubleValue(n)) => {
                assert_eq!(n, 15.0, "Expected sum of 1+2+3+4+5 = 15");
            }
            _ => panic!("Sum should be a number, got {:?}", sum),
        }

        // Cleanup
        for i in 1..=5 {
            let doc_id = format!("doc_{}", i);
            collection
                .document(&doc_id)
                .delete()
                .await
                .expect("Failed to delete document");
        }

        println!("✅ Sum aggregation works!");
    });
}

#[test]
fn test_average_aggregation() {
    SHARED_RUNTIME.block_on(async {
        let firestore = get_firestore().await;
        let unique_id = format!("avg_test_{}", rand::thread_rng().gen::<u32>());
        let collection = firestore.collection(&unique_id);

        // Add test documents
        for i in 1..=4 {
            let doc_id = format!("doc_{}", i);
            let data = create_map(vec![
                ("score", ValueType::IntegerValue(i * 10)),
            ]);

            collection
                .document(&doc_id)
                .set(data)
                .await
                .expect("Failed to set document");
        }

        // Test average aggregation - CollectionReference implements Query trait
        use firebase_rust_sdk::firestore::{AggregateField, Query};
        let avg_field = AggregateField::average("score");
        let aggregate_query = collection.aggregate(vec![avg_field]);

        let snapshot = aggregate_query
            .get()
            .await
            .expect("Failed to execute aggregate query");

        let avg = snapshot
            .get("average_score")
            .expect("Average not found in results");

        match avg.value_type {
            Some(ValueType::DoubleValue(n)) => {
                assert_eq!(n, 25.0, "Expected average of 10+20+30+40 = 25");
            }
            _ => panic!("Average should be a double, got {:?}", avg),
        }

        // Cleanup
        for i in 1..=4 {
            let doc_id = format!("doc_{}", i);
            collection
                .document(&doc_id)
                .delete()
                .await
                .expect("Failed to delete document");
        }

        println!("✅ Average aggregation works!");
    });
}

#[test]
fn test_multiple_aggregations() {
    SHARED_RUNTIME.block_on(async {
        let firestore = get_firestore().await;
        let unique_id = format!("multi_agg_test_{}", rand::thread_rng().gen::<u32>());
        let collection = firestore.collection(&unique_id);

        // Add test documents
        for i in 1..=3 {
            let doc_id = format!("doc_{}", i);
            let data = create_map(vec![
                ("value", ValueType::IntegerValue(i * 100)),
            ]);

            collection
                .document(&doc_id)
                .set(data)
                .await
                .expect("Failed to set document");
        }

        // Test multiple aggregations at once - CollectionReference implements Query trait
        use firebase_rust_sdk::firestore::{AggregateField, Query};
        let aggregations = vec![
            AggregateField::count(),
            AggregateField::sum("value"),
            AggregateField::average("value"),
        ];
        let aggregate_query = collection.aggregate(aggregations);

        let snapshot = aggregate_query
            .get()
            .await
            .expect("Failed to execute aggregate query");

        // Verify count
        let count = snapshot.get("count").expect("Count not found");
        match count.value_type {
            Some(ValueType::IntegerValue(n)) => assert_eq!(n, 3),
            _ => panic!("Count should be an integer"),
        }

        // Verify sum
        let sum = snapshot.get("sum_value").expect("Sum not found");
        match sum.value_type {
            Some(ValueType::IntegerValue(n)) => assert_eq!(n, 600), // 100 + 200 + 300
            Some(ValueType::DoubleValue(n)) => assert_eq!(n, 600.0),
            _ => panic!("Sum should be a number"),
        }

        // Verify average
        let avg = snapshot.get("average_value").expect("Average not found");
        match avg.value_type {
            Some(ValueType::DoubleValue(n)) => assert_eq!(n, 200.0),
            _ => panic!("Average should be a double"),
        }

        // Cleanup
        for i in 1..=3 {
            let doc_id = format!("doc_{}", i);
            collection
                .document(&doc_id)
                .delete()
                .await
                .expect("Failed to delete document");
        }

        println!("✅ Multiple aggregations work!");
    });
}