Asynchronous Programming Patterns for Web Development(0786)

use hyperlane::*;

async fn async_fundamentals_handler(ctx: Context) {
    let start_time = std::time::Instant::now();

    // Demonstrate concurrent async operations
    let (result1, result2, result3) = tokio::join!(
        async_database_operation(),
        async_external_api_call(),
        async_file_operation()
    );

    let total_time = start_time.elapsed();

    // All operations completed concurrently
    let response = format!(r#"{{
        "database_result": "{}",
        "api_result": "{}",
        "file_result": "{}",
        "total_time_ms": {:.3},
        "concurrent_execution": true
    }}"#,
        result1, result2, result3,
        total_time.as_secs_f64() * 1000.0
    );

    ctx.set_response_status_code(200)
        .await
        .set_response_header(CONTENT_TYPE, "application/json")
        .await
        .set_response_header("X-Async-Operations", "3")
        .await
        .set_response_body(response)
        .await;
}

async fn async_database_operation() -> String {
    // Simulate database query
    tokio::time::sleep(tokio::time::Duration::from_millis(100)).await;
    "Database query completed".to_string()
}

async fn async_external_api_call() -> String {
    // Simulate external API call
    tokio::time::sleep(tokio::time::Duration::from_millis(150)).await;
    "External API call completed".to_string()
}

async fn async_file_operation() -> String {
    // Simulate file I/O
    tokio::time::sleep(tokio::time::Duration::from_millis(75)).await;
    "File operation completed".to_string()
}

async fn sequential_vs_concurrent_handler(ctx: Context) {
    // Demonstrate performance difference between sequential and concurrent execution
    let sequential_time = measure_sequential_execution().await;
    let concurrent_time = measure_concurrent_execution().await;

    let performance_improvement = ((sequential_time - concurrent_time) / sequential_time) * 100.0;

    let response = format!(r#"{{
        "sequential_time_ms": {:.3},
        "concurrent_time_ms": {:.3},
        "performance_improvement_percent": {:.1},
        "speedup_factor": {:.2}
    }}"#,
        sequential_time * 1000.0,
        concurrent_time * 1000.0,
        performance_improvement,
        sequential_time / concurrent_time
    );

    ctx.set_response_status_code(200)
        .await
        .set_response_body(response)
        .await;
}

async fn measure_sequential_execution() -> f64 {
    let start = std::time::Instant::now();

    // Sequential execution
    let _ = async_operation(50).await;
    let _ = async_operation(75).await;
    let _ = async_operation(100).await;

    start.elapsed().as_secs_f64()
}

async fn measure_concurrent_execution() -> f64 {
    let start = std::time::Instant::now();

    // Concurrent execution
    let (_, _, _) = tokio::join!(
        async_operation(50),
        async_operation(75),
        async_operation(100)
    );

    start.elapsed().as_secs_f64()
}

async fn async_operation(delay_ms: u64) -> String {
    tokio::time::sleep(tokio::time::Duration::from_millis(delay_ms)).await;
    format!("Operation completed after {}ms", delay_ms)
}

async fn async_stream_processing_handler(ctx: Context) {
    // Demonstrate async stream processing
    ctx.set_response_status_code(200)
        .await
        .set_response_header(CONTENT_TYPE, "application/json")
        .await
        .send()
        .await;

    // Stream processing with async operations
    let _ = ctx.set_response_body("[").await.send_body().await;

    let mut first = true;
    for i in 0..100 {
        // Async processing for each item
        let processed_item = process_stream_item(i).await;

        let json_item = if first {
            first = false;
            format!(r#"{{"id": {}, "data": "{}"}}"#, i, processed_item)
        } else {
            format!(r#",{{"id": {}, "data": "{}"}}"#, i, processed_item)
        };

        if ctx.set_response_body(json_item).await.send_body().await.is_err() {
            break; // Client disconnected
        }

        // Yield control to other tasks
        if i % 10 == 0 {
            tokio::task::yield_now().await;
        }
    }

    let _ = ctx.set_response_body("]").await.send_body().await;
    let _ = ctx.closed().await;
}

async fn process_stream_item(item_id: usize) -> String {
    // Simulate async processing of stream item
    tokio::time::sleep(tokio::time::Duration::from_millis(10)).await;
    format!("Processed item {}", item_id)
}

#[tokio::main]
async fn main() {
    let server: Server = Server::new();
    server.host("0.0.0.0").await;
    server.port(60000).await;

    // Optimize for async operations
    server.enable_nodelay().await;
    server.disable_linger().await;
    server.http_buffer_size(4096).await;

    server.route("/async/fundamentals", async_fundamentals_handler).await;
    server.route("/async/comparison", sequential_vs_concurrent_handler).await;
    server.route("/async/stream", async_stream_processing_handler).await;

    server.run().await.unwrap();
}

async fn async_pipeline_handler(ctx: Context) {
    // Demonstrate async processing pipeline
    let input_data = ctx.get_request_body().await;

    // Create async processing pipeline
    let pipeline_result = execute_async_pipeline(&input_data).await;

    match pipeline_result {
        Ok(result) => {
            ctx.set_response_status_code(200)
                .await
                .set_response_body(result)
                .await;
        }
        Err(e) => {
            ctx.set_response_status_code(500)
                .await
                .set_response_body(format!("Pipeline error: {}", e))
                .await;
        }
    }
}

async fn execute_async_pipeline(input: &[u8]) -> Result<String, Box<dyn std::error::Error>> {
    // Stage 1: Validation
    let validated_data = validate_input_async(input).await?;

    // Stage 2: Transformation
    let transformed_data = transform_data_async(&validated_data).await?;

    // Stage 3: Enrichment (concurrent operations)
    let (enriched_data, metadata) = tokio::try_join!(
        enrich_data_async(&transformed_data),
        fetch_metadata_async(&transformed_data)
    )?;

    // Stage 4: Finalization
    let final_result = finalize_data_async(&enriched_data, &metadata).await?;

    Ok(final_result)
}

async fn validate_input_async(input: &[u8]) -> Result<String, Box<dyn std::error::Error>> {
    // Async validation
    tokio::time::sleep(tokio::time::Duration::from_millis(10)).await;

    if input.is_empty() {
        return Err("Empty input".into());
    }

    Ok(String::from_utf8_lossy(input).to_string())
}

async fn transform_data_async(data: &str) -> Result<String, Box<dyn std::error::Error>> {
    // Async transformation
    tokio::time::sleep(tokio::time::Duration::from_millis(20)).await;
    Ok(data.to_uppercase())
}

async fn enrich_data_async(data: &str) -> Result<String, Box<dyn std::error::Error>> {
    // Async data enrichment
    tokio::time::sleep(tokio::time::Duration::from_millis(30)).await;
    Ok(format!("ENRICHED: {}", data))
}

async fn fetch_metadata_async(data: &str) -> Result<String, Box<dyn std::error::Error>> {
    // Async metadata fetching
    tokio::time::sleep(tokio::time::Duration::from_millis(25)).await;
    Ok(format!("META: length={}", data.len()))
}

async fn finalize_data_async(data: &str, metadata: &str) -> Result<String, Box<dyn std::error::Error>> {
    // Async finalization
    tokio::time::sleep(tokio::time::Duration::from_millis(15)).await;

    Ok(format!(r#"{{
        "data": "{}",
        "metadata": "{}",
        "timestamp": {}
    }}"#, data, metadata, current_timestamp()))
}

fn current_timestamp() -> u64 {
    std::time::SystemTime::now()
        .duration_since(std::time::UNIX_EPOCH)
        .unwrap()
        .as_secs()
}

async fn async_error_handling_handler(ctx: Context) {
    // Demonstrate async error handling patterns
    let error_scenarios = vec![
        "success",
        "timeout",
        "network_error",
        "validation_error",
    ];

    let mut results = Vec::new();

    for scenario in error_scenarios {
        let result = handle_async_scenario(scenario).await;
        results.push(format!(r#"{{"scenario": "{}", "result": "{}"}}"#, scenario, result));
    }

    let response = format!("[{}]", results.join(","));

    ctx.set_response_status_code(200)
        .await
        .set_response_body(response)
        .await;
}

async fn handle_async_scenario(scenario: &str) -> String {
    match execute_async_scenario(scenario).await {
        Ok(result) => format!("Success: {}", result),
        Err(e) => format!("Error: {}", e),
    }
}

async fn execute_async_scenario(scenario: &str) -> Result<String, Box<dyn std::error::Error>> {
    match scenario {
        "success" => {
            tokio::time::sleep(tokio::time::Duration::from_millis(50)).await;
            Ok("Operation completed successfully".to_string())
        }
        "timeout" => {
            // Simulate timeout with race condition
            let operation = tokio::time::sleep(tokio::time::Duration::from_millis(1000));
            let timeout = tokio::time::sleep(tokio::time::Duration::from_millis(100));

            tokio::select! {
                _ = operation => Ok("Operation completed".to_string()),
                _ = timeout => Err("Operation timed out".into()),
            }
        }
        "network_error" => {
            tokio::time::sleep(tokio::time::Duration::from_millis(30)).await;
            Err("Network connection failed".into())
        }
        "validation_error" => {
            tokio::time::sleep(tokio::time::Duration::from_millis(10)).await;
            Err("Input validation failed".into())
        }
        _ => Err("Unknown scenario".into()),
    }
}

async fn async_concurrency_control_handler(ctx: Context) {
    // Demonstrate concurrency control patterns
    let concurrency_results = test_concurrency_patterns().await;

    ctx.set_response_status_code(200)
        .await
        .set_response_body(concurrency_results)
        .await;
}

async fn test_concurrency_patterns() -> String {
    // Test different concurrency control mechanisms
    let semaphore_result = test_semaphore_pattern().await;
    let rate_limit_result = test_rate_limiting_pattern().await;
    let circuit_breaker_result = test_circuit_breaker_pattern().await;

    format!(r#"{{
        "semaphore_test": "{}",
        "rate_limit_test": "{}",
        "circuit_breaker_test": "{}"
    }}"#, semaphore_result, rate_limit_result, circuit_breaker_result)
}

async fn test_semaphore_pattern() -> String {
    // Limit concurrent operations using semaphore
    let semaphore = std::sync::Arc::new(tokio::sync::Semaphore::new(3)); // Max 3 concurrent

    let mut tasks = Vec::new();

    for i in 0..10 {
        let sem = semaphore.clone();
        let task = tokio::spawn(async move {
            let _permit = sem.acquire().await.unwrap();

            // Simulate work
            tokio::time::sleep(tokio::time::Duration::from_millis(100)).await;
            format!("Task {} completed", i)
        });

        tasks.push(task);
    }

    let results: Vec<_> = futures::future::join_all(tasks).await;
    format!("Completed {} tasks with semaphore control", results.len())
}

async fn test_rate_limiting_pattern() -> String {
    // Implement rate limiting using token bucket
    let rate_limiter = std::sync::Arc::new(tokio::sync::Semaphore::new(5)); // 5 tokens

    let mut successful_operations = 0;

    for _ in 0..10 {
        if let Ok(_permit) = rate_limiter.try_acquire() {
            // Operation allowed
            tokio::time::sleep(tokio::time::Duration::from_millis(10)).await;
            successful_operations += 1;
        }
        // Rate limited operations are dropped
    }

    format!("Rate limiter allowed {} out of 10 operations", successful_operations)
}

async fn test_circuit_breaker_pattern() -> String {
    // Simple circuit breaker implementation
    let mut failure_count = 0;
    let mut successful_operations = 0;
    let failure_threshold = 3;

    for i in 0..10 {
        if failure_count >= failure_threshold {
            // Circuit breaker is open
            continue;
        }

        // Simulate operation that might fail
        if i % 4 == 0 {
            // Simulate failure
            failure_count += 1;
        } else {
            // Simulate success
            successful_operations += 1;
            failure_count = 0; // Reset on success
        }

        tokio::time::sleep(tokio::time::Duration::from_millis(10)).await;
    }

    format!("Circuit breaker allowed {} successful operations", successful_operations)
}

async fn async_performance_handler(ctx: Context) {
    let performance_analysis = analyze_async_performance().await;

    ctx.set_response_status_code(200)
        .await
        .set_response_header("X-Performance-Analysis", "complete")
        .await
        .set_response_body(performance_analysis)
        .await;
}

async fn analyze_async_performance() -> String {
    let task_spawn_overhead = measure_task_spawn_overhead().await;
    let async_fn_overhead = measure_async_fn_overhead().await;
    let context_switch_overhead = measure_context_switch_overhead().await;
    let memory_usage = measure_async_memory_usage().await;

    format!(r#"{{
        "task_spawn_overhead_ns": {:.2},
        "async_fn_overhead_ns": {:.2},
        "context_switch_overhead_ns": {:.2},
        "memory_usage_per_task_bytes": {},
        "recommendations": "Use async for I/O-bound operations, avoid for CPU-bound tasks"
    }}"#,
        task_spawn_overhead,
        async_fn_overhead,
        context_switch_overhead,
        memory_usage
    )
}

async fn measure_task_spawn_overhead() -> f64 {
    let iterations = 10000;
    let start = std::time::Instant::now();

    let mut tasks = Vec::new();
    for _ in 0..iterations {
        let task = tokio::spawn(async {
            // Minimal work
            42
        });
        tasks.push(task);
    }

    // Wait for all tasks to complete
    for task in tasks {
        let _ = task.await;
    }

    let elapsed = start.elapsed();
    elapsed.as_nanos() as f64 / iterations as f64
}

async fn measure_async_fn_overhead() -> f64 {
    let iterations = 100000;
    let start = std::time::Instant::now();

    for _ in 0..iterations {
        let _ = minimal_async_function().await;
    }

    let elapsed = start.elapsed();
    elapsed.as_nanos() as f64 / iterations as f64
}

async fn minimal_async_function() -> i32 {
    42
}

async fn measure_context_switch_overhead() -> f64 {
    let iterations = 10000;
    let start = std::time::Instant::now();

    for _ in 0..iterations {
        tokio::task::yield_now().await;
    }

    let elapsed = start.elapsed();
    elapsed.as_nanos() as f64 / iterations as f64
}

async fn measure_async_memory_usage() -> usize {
    // Estimate memory usage per async task
    let base_memory = get_memory_usage();

    let mut tasks = Vec::new();
    for _ in 0..1000 {
        let task = tokio::spawn(async {
            // Hold some state
            let _data = vec![0u8; 1024]; // 1KB per task
            tokio::time::sleep(tokio::time::Duration::from_millis(100)).await;
        });
        tasks.push(task);
    }

    let peak_memory = get_memory_usage();

    // Wait for tasks to complete
    for task in tasks {
        let _ = task.await;
    }

    (peak_memory - base_memory) / 1000 // Per task
}

fn get_memory_usage() -> usize {
    // Simulate memory usage measurement
    1024 * 1024 * 50 // 50MB baseline
}

async fn async_best_practices_handler(ctx: Context) {
    let best_practices_demo = demonstrate_async_best_practices().await;

    ctx.set_response_status_code(200)
        .await
        .set_response_body(best_practices_demo)
        .await;
}

async fn demonstrate_async_best_practices() -> String {
    let mut practices = Vec::new();

    // Practice 1: Avoid blocking operations in async context
    let practice1_result = avoid_blocking_operations().await;
    practices.push(format!("Avoid blocking: {}", practice1_result));

    // Practice 2: Use appropriate concurrency patterns
    let practice2_result = use_appropriate_concurrency().await;
    practices.push(format!("Appropriate concurrency: {}", practice2_result));

    // Practice 3: Handle errors gracefully
    let practice3_result = handle_errors_gracefully().await;
    practices.push(format!("Error handling: {}", practice3_result));

    // Practice 4: Optimize for the common case
    let practice4_result = optimize_common_case().await;
    practices.push(format!("Optimization: {}", practice4_result));

    format!("Async Best Practices:\n{}", practices.join("\n"))
}

async fn avoid_blocking_operations() -> String {
    // Good: Use async I/O
    let start = std::time::Instant::now();
    tokio::time::sleep(tokio::time::Duration::from_millis(10)).await;
    let async_time = start.elapsed();

    // Demonstrate why blocking is bad (simulated)
    let blocking_time = std::time::Duration::from_millis(10);

    format!("Async I/O: {:.3}ms, Blocking would be: {:.3}ms",
            async_time.as_secs_f64() * 1000.0,
            blocking_time.as_secs_f64() * 1000.0)
}

async fn use_appropriate_concurrency() -> String {
    // Use join! for independent operations
    let start = std::time::Instant::now();

    let (result1, result2) = tokio::join!(
        async { tokio::time::sleep(tokio::time::Duration::from_millis(20)).await; "Task 1" },
        async { tokio::time::sleep(tokio::time::Duration::from_millis(30)).await; "Task 2" }
    );

    let concurrent_time = start.elapsed();

    format!("Concurrent execution of {} and {} took {:.3}ms",
            result1, result2, concurrent_time.as_secs_f64() * 1000.0)
}

async fn handle_errors_gracefully() -> String {
    // Use Result types and proper error propagation
    match risky_async_operation().await {
        Ok(result) => format!("Success: {}", result),
        Err(e) => format!("Handled error: {}", e),
    }
}

async fn risky_async_operation() -> Result<String, Box<dyn std::error::Error>> {
    tokio::time::sleep(tokio::time::Duration::from_millis(5)).await;

    if rand::random::<f32>() < 0.5 {
        Ok("Operation succeeded".to_string())
    } else {
        Err("Simulated error".into())
    }
}

async fn optimize_common_case() -> String {
    // Optimize for the most common execution path
    let start = std::time::Instant::now();

    // Fast path for common case
    let result = if is_common_case() {
        handle_common_case().await
    } else {
        handle_rare_case().await
    };

    let execution_time = start.elapsed();

    format!("Optimized execution: {} in {:.3}ms",
            result, execution_time.as_secs_f64() * 1000.0)
}

fn is_common_case() -> bool {
    rand::random::<f32>() < 0.8 // 80% common case
}

async fn handle_common_case() -> String {
    // Fast path
    tokio::time::sleep(tokio::time::Duration::from_millis(5)).await;
    "Common case handled quickly".to_string()
}

async fn handle_rare_case() -> String {
    // Slower path for rare cases
    tokio::time::sleep(tokio::time::Duration::from_millis(50)).await;
    "Rare case handled thoroughly".to_string()
}

async fn real_world_async_handler(ctx: Context) {
    // Simulate real-world async application
    let application_result = simulate_real_world_application(&ctx).await;

    ctx.set_response_status_code(200)
        .await
        .set_response_body(application_result)
        .await;
}

async fn simulate_real_world_application(ctx: &Context) -> String {
    let start_time = std::time::Instant::now();

    // Simulate complex application workflow
    let user_id = ctx.get_route_param("user_id").await.unwrap_or_default();

    // Concurrent operations typical in real applications
    let (user_data, permissions, preferences, audit_log) = tokio::join!(
        fetch_user_data(&user_id),
        fetch_user_permissions(&user_id),
        fetch_user_preferences(&user_id),
        log_user_access(&user_id)
    );

    // Process results
    let processed_data = process_user_context(user_data, permissions, preferences).await;

    let total_time = start_time.elapsed();

    format!(r#"{{
        "user_id": "{}",
        "processed_data": "{}",
        "audit_logged": {},
        "processing_time_ms": {:.3},
        "async_operations": 4
    }}"#,
        user_id,
        processed_data,
        audit_log,
        total_time.as_secs_f64() * 1000.0
    )
}

async fn fetch_user_data(user_id: &str) -> String {
    // Simulate database query
    tokio::time::sleep(tokio::time::Duration::from_millis(50)).await;
    format!("UserData for {}", user_id)
}

async fn fetch_user_permissions(user_id: &str) -> String {
    // Simulate permissions lookup
    tokio::time::sleep(tokio::time::Duration::from_millis(30)).await;
    format!("Permissions for {}", user_id)
}

async fn fetch_user_preferences(user_id: &str) -> String {
    // Simulate preferences retrieval
    tokio::time::sleep(tokio::time::Duration::from_millis(20)).await;
    format!("Preferences for {}", user_id)
}

async fn log_user_access(user_id: &str) -> bool {
    // Simulate audit logging
    tokio::time::sleep(tokio::time::Duration::from_millis(10)).await;
    println!("User {} accessed the system", user_id);
    true
}

async fn process_user_context(data: String, permissions: String, preferences: String) -> String {
    // Process combined user context
    tokio::time::sleep(tokio::time::Duration::from_millis(15)).await;
    format!("Processed: {} | {} | {}", data, permissions, preferences)
}