use std::collections::HashMap;
use std::net::SocketAddr;
use std::pin::Pin;
use std::sync::Arc;
use std::time::Duration;

use anyhow::Result;
use futures::StreamExt;
use grpc_graphql_gateway::{
    Gateway, GrpcClient, HeaderPropagationConfig,
    HighPerfConfig, RateLimitMiddleware,
};
use tokio::fs;
use tokio::sync::RwLock;
use tokio_stream::wrappers::IntervalStream;
use tokio_stream::Stream;
use tonic::{transport::Server, Request, Response, Status};
use tracing::info;

pub mod greeter {
    include!("../../src/generated/greeter.rs");
}

use greeter::greeter_server::{Greeter, GreeterServer};
use greeter::{
    GetUserRequest, GreetMeta, HelloReply, HelloRequest, UpdateGreetingRequest, UploadAvatarReply,
    UploadAvatarRequest, UploadAvatarsReply, UploadAvatarsRequest, User,
};

const DESCRIPTORS: &[u8] = include_bytes!("../../src/generated/greeter_descriptor.bin");
const GRPC_ADDR: &str = "127.0.0.1:50051";
const GQL_ADDR: &str = "127.0.0.1:8888";

#[tokio::main]
async fn main() -> Result<()> {
    tracing_subscriber::fmt()
        .with_env_filter(
            tracing_subscriber::EnvFilter::try_from_default_env()
                .unwrap_or_else(|_| tracing_subscriber::EnvFilter::new("info")),
        )
        .compact()
        .init();

    let grpc_addr: SocketAddr = GRPC_ADDR.parse()?;
    let gql_addr: SocketAddr = GQL_ADDR.parse()?;

    let greeter = ExampleGreeter::default();

    print_examples(gql_addr);

    let grpc = tokio::spawn(run_grpc_server(greeter.clone(), grpc_addr));
    let gateway = tokio::spawn(run_gateway(gql_addr));

    let (grpc_res, gateway_res) = tokio::join!(grpc, gateway);
    grpc_res??;
    gateway_res??;

    Ok(())
}

async fn run_grpc_server(greeter: ExampleGreeter, addr: SocketAddr) -> Result<()> {
    info!("gRPC Greeter listening on {}", addr);

    Server::builder()
        .add_service(GreeterServer::new(greeter))
        .serve(addr)
        .await?;

    Ok(())
}

async fn run_gateway(addr: SocketAddr) -> Result<()> {
    info!(
        "GraphQL gateway listening on http://{}/graphql (ws://{}/graphql/ws for subscriptions)",
        addr, addr
    );

    let client = GrpcClient::builder(format!("http://{GRPC_ADDR}"))
        .lazy(false)
        .connect()
        .await?;

    // Configure a REST connector from OpenAPI spec (JSONPlaceholder API)
    // This demonstrates the OpenAPI integration feature
    // We embed the YAML at compile time to avoid path issues
    const OPENAPI_YAML: &str = include_str!("jsonplaceholder.yaml");

    let jsonplaceholder_api = grpc_graphql_gateway::OpenApiParser::from_string(OPENAPI_YAML, true)
        .expect("Failed to parse OpenAPI YAML")
        .with_prefix("jp_") // Prefix all operations with jp_ to namespace them
        .with_tags(vec!["posts".to_string(), "users".to_string()]) // Only posts and users
        .build()
        .expect("Failed to build REST connector from OpenAPI spec");

    info!("REST connector configured from OpenAPI spec (jsonplaceholder.yaml)");
    info!("Available operations:");
    for (name, endpoint) in jsonplaceholder_api.endpoints() {
        info!("  - {}: {:?} {}", name, endpoint.method, endpoint.path);
    }

    Gateway::builder()
        .with_descriptor_set_bytes(DESCRIPTORS)
        .add_grpc_client("greeter.Greeter", client)
        // Add REST connector for external API integration
        .add_rest_connector("jsonplaceholder", jsonplaceholder_api)
        // Add a rate limiter: 10 req/s with burst of 5
        .add_middleware(RateLimitMiddleware::new(10, 5))
        .with_high_performance(HighPerfConfig::ultra_fast())
        // Add enhanced structured logging middleware
        /*
        .add_middleware(EnhancedLoggingMiddleware::new(
            LoggingConfig::default()
                .with_level(LogLevel::Info)
                .with_headers(true)
                .mask_header("x-secret-key")  // Custom sensitive header
        ))
        */
        // Enable APQ for bandwidth optimization
        .with_persisted_queries(grpc_graphql_gateway::PersistedQueryConfig {
            cache_size: 100,
            ttl: None, // No expiration
        })
        // Enable Circuit Breaker for resilience
        .with_circuit_breaker(grpc_graphql_gateway::CircuitBreakerConfig {
            failure_threshold: 3,                      // Open after 3 failures
            recovery_timeout: Duration::from_secs(10), // Test recovery after 10s
            half_open_max_requests: 2,                 // Allow 2 test requests
        })
        // Enable Response Caching for performance
        .with_response_cache(grpc_graphql_gateway::CacheConfig {
            max_size: 10000,                        // Increase max size for benchmark
            default_ttl: Duration::from_secs(3600), // Longer TTL for benchmark
            stale_while_revalidate: Some(Duration::from_secs(30)),
            invalidate_on_mutation: true,
            redis_url: None, // Use in-memory ShardedCache for max speed
            vary_headers: vec!["Authorization".to_string()],
            smart_ttl_manager: None, // Optional: Add SmartTtlManager for intelligent TTL
        })
        // Enable Response Compression for bandwidth savings
        .with_compression(grpc_graphql_gateway::CompressionConfig::default())
        // Enable Header Propagation for auth and tracing
        .with_header_propagation(
            HeaderPropagationConfig::new()
                .propagate("authorization")
                .propagate("x-request-id")
                .propagate("x-tenant-id")
                .propagate_with_prefix("x-custom-"),
        )
        // Enable Query Whitelisting for production security
        .with_query_whitelist(
            grpc_graphql_gateway::QueryWhitelistConfig::from_json_file(
                "examples/allowed_queries.json",
                grpc_graphql_gateway::WhitelistMode::Warn,
            )
            .map(|mut c| {
                c.allow_introspection = true;
                c
            })
            .unwrap_or_else(|e| {
                eprintln!(
                    "Warning: Failed to load query whitelist: {}. Using empty whitelist.",
                    e
                );
                grpc_graphql_gateway::QueryWhitelistConfig::warn()
            }),
        )
        // Enable Query Analytics Dashboard
        // .enable_analytics(AnalyticsConfig::development())
        // Enable Request Collapsing for deduplicating identical gRPC calls
        .with_request_collapsing(grpc_graphql_gateway::RequestCollapsingConfig::default())
        // Enable Prometheus Metrics
        .enable_metrics()
        .serve(addr.to_string())
        .await?;

    Ok(())
}

fn print_examples(addr: SocketAddr) {
    println!("GraphQL endpoint: http://{}/graphql", addr);
    println!("WebSocket endpoint: ws://{}/graphql/ws", addr);
    // curl (query): curl -X POST http://127.0.0.1:8888/graphql -H 'content-type: application/json' -d '{"query":"{ hello(name:\"GraphQL\"){ message } }"}'
    // curl (mutation): curl -X POST http://127.0.0.1:8888/graphql -H 'content-type: application/json' -d '{"query":"mutation { updateGreeting(input:{ name:\"GraphQL\", salutation:\"Howdy\" }) { message } }"}'
    // curl (resolver): curl -X POST http://127.0.0.1:8888/graphql -H 'content-type: application/json' -d '{"query":"{ user(id:\"demo\"){ id displayName trusted } }"}'
    // subscription (graphql-transport-ws):
    //   websocat -H="Sec-WebSocket-Protocol: graphql-transport-ws" --protocol graphql-transport-ws ws://127.0.0.1:8888/graphql/ws
    //   # then type/paste:
    //   {"type":"connection_init","payload":{}}
    //   {"id":"1","type":"subscribe","payload":{"query":"subscription { streamHello(name:\"GraphQL\"){ message } }"}}
    //   # server replies with {"type":"connection_ack"} then pushes data payloads.
    println!("Try these operations once the servers are up:");
    println!();
    println!("=== gRPC Backend (via GraphQL) ===");
    println!(
        "  query {{ hello(name:\"GraphQL\") {{ message meta {{ correlationId from {{ id displayName trusted }} }} }} }}"
    );
    println!(
        "  mutation {{ updateGreeting(input:{{ name:\"GraphQL\", salutation:\"Howdy\" }}) {{ message }} }}"
    );
    println!(
        "  subscription {{ streamHello(name:\"GraphQL\") {{ message meta {{ correlationId }} }} }}"
    );
    println!("  query {{ user(id:\"demo\") {{ id displayName trusted }} }}");
    println!();
    println!("=== REST Connector (JSONPlaceholder API) ===");
    println!("The gateway includes a REST connector for https://jsonplaceholder.typicode.com");
    println!("You can execute REST endpoints programmatically via the connector:");
    println!();
    println!("  // In your resolver or middleware:");
    println!("  let connector = gateway.rest_connectors().get(\"jsonplaceholder\").unwrap();");
    println!("  let result = connector.execute(\"getPost\", args).await?;");
    println!();
    println!("Test the REST connector directly with curl:");
    println!("  curl https://jsonplaceholder.typicode.com/posts/1");
    println!("  curl https://jsonplaceholder.typicode.com/users/1");
    println!();
    println!("=== File Uploads ===");
    println!("  # Upload (multipart): see README for the curl example");
    println!("  # Multi-upload (multipart): see README for the curl example");
    println!();
    println!("NOTE: Query whitelisting is enabled in WARN mode.");
    println!("      Non-whitelisted queries will log warnings but still execute.");
    println!("      Set WhitelistMode::Enforce in production for strict security!");
    println!();
    println!("NOTE: REST connector is configured for JSONPlaceholder API demo.");
    println!("      Use add_rest_connector() to integrate your own REST APIs!");
    println!();
    println!("=== Query Analytics Dashboard ===");
    println!("  Dashboard UI: http://{}/analytics", addr);
    println!("  JSON API:     http://{}/analytics/api", addr);
    println!(
        "  Reset:        curl -X POST http://{}/analytics/reset",
        addr
    );
}

#[derive(Clone)]
struct ExampleGreeter {
    greeting: Arc<RwLock<String>>,
    users: Arc<RwLock<HashMap<String, User>>>,
}

impl Default for ExampleGreeter {
    fn default() -> Self {
        let mut users = HashMap::new();
        users.insert(
            "demo".to_string(),
            User {
                id: "demo".to_string(),
                display_name: "Demo User".to_string(),
                trusted: true,
            },
        );
        users.insert(
            "admin".to_string(),
            User {
                id: "admin".to_string(),
                display_name: "Admin User".to_string(),
                trusted: false,
            },
        );

        Self {
            greeting: Arc::new(RwLock::new("Hello".to_string())),
            users: Arc::new(RwLock::new(users)),
        }
    }
}

#[tonic::async_trait]
impl Greeter for ExampleGreeter {
    async fn say_hello(
        &self,
        request: Request<HelloRequest>,
    ) -> Result<Response<HelloReply>, Status> {
        // Log any propagated headers/metadata
        let metadata = request.metadata();
        if let Some(auth) = metadata.get("authorization") {
            info!("Received Authorization header: {:?}", auth);
        }
        if let Some(req_id) = metadata.get("x-request-id") {
            info!("Received X-Request-ID: {:?}", req_id);
        }
        if let Some(tenant) = metadata.get("x-tenant-id") {
            info!("Received X-Tenant-ID: {:?}", tenant);
        }

        let req = request.into_inner();
        let reply = self.build_reply(normalize_name(req.name), "query").await;
        Ok(Response::new(reply))
    }

    async fn update_greeting(
        &self,
        request: Request<UpdateGreetingRequest>,
    ) -> Result<Response<HelloReply>, Status> {
        let req = request.into_inner();
        {
            let mut greeting = self.greeting.write().await;
            *greeting = req.greeting.clone();
        }

        let reply = self.build_reply(normalize_name(req.name), "mutation").await;
        Ok(Response::new(reply))
    }

    type StreamHellosStream =
        Pin<Box<dyn Stream<Item = Result<HelloReply, Status>> + Send + 'static>>;

    async fn stream_hellos(
        &self,
        request: Request<HelloRequest>,
    ) -> Result<Response<Self::StreamHellosStream>, Status> {
        let name = normalize_name(request.into_inner().name);
        let greeting = self.greeting.clone();
        let users = self.users.clone();

        let stream = IntervalStream::new(tokio::time::interval(Duration::from_secs(1)))
            .enumerate()
            .take(5)
            .then(move |(idx, _)| {
                let greeting = greeting.clone();
                let users = users.clone();
                let name = name.clone();
                async move {
                    let greeting = greeting.read().await.clone();
                    let user = users.read().await.get("demo").cloned();

                    Ok(HelloReply {
                        message: format!("{greeting}, {name}! (#{})", idx + 1),
                        meta: Some(GreetMeta {
                            correlation_id: format!("stream-{idx}"),
                            from: user,
                        }),
                    })
                }
            });

        Ok(Response::new(Box::pin(stream) as Self::StreamHellosStream))
    }

    async fn resolve_user(
        &self,
        request: Request<GetUserRequest>,
    ) -> Result<Response<User>, Status> {
        let req = request.into_inner();
        if let Some(user) = self.lookup_user(&req.id).await {
            return Ok(Response::new(user));
        }

        Err(Status::not_found(format!("user {} not found", req.id)))
    }

    async fn upload_avatar(
        &self,
        request: Request<UploadAvatarRequest>,
    ) -> Result<Response<UploadAvatarReply>, Status> {
        let req = request.into_inner();
        if self.lookup_user(&req.user_id).await.is_none() {
            return Err(Status::not_found(format!("user {} not found", req.user_id)));
        }

        let mut path = std::env::temp_dir();
        path.push(format!(
            "greeter_avatar_{}.bin",
            safe_filename(&req.user_id)
        ));
        fs::write(&path, &req.avatar)
            .await
            .map_err(|e| Status::internal(format!("failed to write avatar: {e}")))?;
        info!("stored avatar for {} at {}", req.user_id, path.display());

        let size = req.avatar.len() as u64;
        let reply = UploadAvatarReply {
            user_id: req.user_id,
            size,
        };
        Ok(Response::new(reply))
    }

    async fn upload_avatars(
        &self,
        request: Request<UploadAvatarsRequest>,
    ) -> Result<Response<UploadAvatarsReply>, Status> {
        let req = request.into_inner();
        if self.lookup_user(&req.user_id).await.is_none() {
            return Err(Status::not_found(format!("user {} not found", req.user_id)));
        }

        let mut sizes = Vec::with_capacity(req.avatars.len());
        for (idx, blob) in req.avatars.iter().enumerate() {
            let mut path = std::env::temp_dir();
            path.push(format!(
                "greeter_avatar_{}_{}.bin",
                safe_filename(&req.user_id),
                idx
            ));
            fs::write(&path, blob)
                .await
                .map_err(|e| Status::internal(format!("failed to write avatar: {e}")))?;
            info!(
                "stored avatar {} for {} at {}",
                idx,
                req.user_id,
                path.display()
            );
            sizes.push(blob.len() as u64);
        }

        let reply = UploadAvatarsReply {
            user_id: req.user_id,
            sizes,
        };
        Ok(Response::new(reply))
    }
}

impl ExampleGreeter {
    async fn build_reply(&self, name: String, suffix: &str) -> HelloReply {
        let greeting = self.greeting.read().await.clone();
        let user = self.lookup_user("demo").await;

        HelloReply {
            message: format!("{greeting}, {name}!"),
            meta: Some(GreetMeta {
                correlation_id: format!("hello-{suffix}-{name}"),
                from: user,
            }),
        }
    }

    async fn lookup_user(&self, id: &str) -> Option<User> {
        self.users.read().await.get(id).cloned()
    }
}

fn normalize_name(name: String) -> String {
    if name.trim().is_empty() {
        "World".to_string()
    } else {
        name
    }
}

fn safe_filename(input: &str) -> String {
    input
        .chars()
        .map(|c| if c.is_ascii_alphanumeric() { c } else { '_' })
        .collect()
}
