use std::sync::Arc;
use std::time::Instant;

use actix_web::dev::ServiceResponse;
use actix_web::test;
use actix_web::web::Data;
use actix_web::{App, http::StatusCode};
use athena_rs::AppState;
use athena_rs::api::gateway::insert::{InsertWindowCoordinator, InsertWindowSettings};
use athena_rs::api::metrics::{ClusterProbeMetric, MetricsState, prometheus_metrics};
use athena_rs::drivers::postgresql::deadpool_registry::DeadpoolPostgresRegistry;
use athena_rs::drivers::postgresql::sqlx_driver::PostgresClientRegistry;
use axum::body::Bytes;
use criterion::{Criterion, black_box, criterion_group, criterion_main};
use moka::future::Cache;
use reqwest::Client;
use serde_json::Value;
use sqlx::{Pool, Postgres};
use tokio::runtime::Runtime;

fn build_state(seed_cardinality: usize) -> Data<AppState> {
    let cache: Arc<Cache<String, Value>> = Arc::new(Cache::builder().build());
    let immortal_cache: Arc<Cache<String, Value>> = Arc::new(Cache::builder().build());
    let client: Client = Client::builder().build().expect("http client");
    let pg_registry: Arc<PostgresClientRegistry> = Arc::new(PostgresClientRegistry::empty());

    let jdbc_pool_cache: Arc<Cache<String, Pool<Postgres>>> =
        Arc::new(Cache::builder().max_capacity(512).build());

    let metrics_state: Arc<MetricsState> = Arc::new(MetricsState::new());

    for index in 0..seed_cardinality {
        let route: &str = if index % 3 == 0 {
            "/gateway/data"
        } else if index % 3 == 1 {
            "/management/clients"
        } else {
            "/schema/tables"
        };
        let method: &str = if index % 2 == 0 { "GET" } else { "POST" };
        let status = match index % 5 {
            0 => 200,
            1 => 201,
            2 => 400,
            3 => 404,
            _ => 500,
        };
        let duration_seconds: f64 = (index % 250) as f64 / 1000.0;
        let request_bytes: Option<u64> = Some((512 + (index % 4096)) as u64);
        let response_bytes: Option<u64> = Some((1024 + (index % 8192)) as u64);
        let client_name: Option<&str> = if index % 2 == 0 {
            Some("athena_logging")
        } else {
            Some("reporting")
        };

        metrics_state.begin_http_request(method, route);
        metrics_state.finish_http_request(
            method,
            route,
            status,
            duration_seconds,
            request_bytes,
            response_bytes,
            client_name,
        );

        if index % 11 == 0 {
            metrics_state.record_http_handler_error(
                method,
                route,
                duration_seconds,
                request_bytes,
                client_name,
            );
        }

        metrics_state.record_management_mutation(
            if index % 2 == 0 {
                "create_table"
            } else {
                "drop_table"
            },
            if index % 4 == 0 { "success" } else { "error" },
            duration_seconds,
        );

        metrics_state.set_cluster_probe(
            &format!("https://mirror-{}.athena.local", index % 16),
            ClusterProbeMetric {
                up: index % 7 != 0,
                latency_ms: Some((5 + (index % 200)) as f64),
                download_bytes_per_sec: Some((1_000_000 + (index * 177) % 900_000) as f64),
            },
        );
    }

    let insert_window_coordinator: Arc<InsertWindowCoordinator> =
        InsertWindowCoordinator::new(InsertWindowSettings {
            max_batch: 100,
            max_queued: 10_000,
            deny_tables: Default::default(),
        });
    Data::new(AppState {
        cache,
        immortal_cache,
        client,
        process_start_time_seconds: 0,
        process_started_at: Instant::now(),
        pg_registry,
        jdbc_pool_cache,
        gateway_force_camel_case_to_snake_case: false,
        gateway_auto_cast_uuid_filter_values_to_text: true,
        gateway_allow_schema_names_prefixed_as_table_name: true,
        pipeline_registry: None,
        logging_client_name: None,
        gateway_auth_client_name: None,
        gateway_benchmark_client_name: None,
        gateway_jdbc_allow_private_hosts: false,
        gateway_jdbc_allowed_hosts: vec!["localhost".to_string()],
        gateway_resilience_timeout_secs: 30,
        gateway_resilience_read_max_retries: 1,
        gateway_resilience_initial_backoff_ms: 100,
        prometheus_metrics_enabled: true,
        metrics_state,
        deadpool_registry: Arc::new(DeadpoolPostgresRegistry::empty()),
        #[cfg(feature = "deadpool_experimental")]
        jdbc_deadpool_cache: Arc::new(Cache::builder().max_capacity(512).build()),
        gateway_api_key_fail_mode: "fail_closed".to_string(),
        gateway_admission_store_backend: "redis".to_string(),
        gateway_admission_store_fail_mode: "fail_closed".to_string(),
        gateway_insert_execution_window_ms: 0,
        gateway_insert_window_max_batch: 100,
        gateway_insert_window_max_queued: 10_000,
        gateway_insert_merge_deny_tables: Default::default(),
        insert_window_coordinator,
        client_stats_batcher: None,
        gateway_log_batcher: None,
        logging_task_limiter: None,
    })
}

fn bench_metrics_export(c: &mut Criterion) {
    let state: Data<AppState> = build_state(5_000);
    let runtime: Runtime = Runtime::new().expect("tokio runtime");
    let app = runtime.block_on(test::init_service(
        App::new()
            .app_data(state.clone())
            .service(prometheus_metrics),
    ));

    c.bench_function("prometheus_metrics_export_5000", |b| {
        b.iter(|| {
            let req = test::TestRequest::get().uri("/metrics").to_request();
            let response: ServiceResponse = runtime.block_on(test::call_service(&app, req));
            assert_eq!(response.status(), StatusCode::OK);
            let body_bytes: Bytes = runtime.block_on(test::read_body(response));
            black_box(body_bytes.len());
        });
    });
}

criterion_group!(benches, bench_metrics_export);
criterion_main!(benches);
