use actix_web::{
    HttpRequest, HttpResponse, Responder, delete, get, patch, post, put,
    web::{self, Data, Json, Path},
};
use chrono::{DateTime, Utc};
use serde::Deserialize;
use serde_json::json;
use sha256::digest;
use sqlx::postgres::PgPool;
use uuid::Uuid;

use crate::AppState;
use crate::api::auth::authorize_static_admin_key;
use crate::api::response::{
    api_created, api_success, bad_request, conflict, internal_error, not_found, service_unavailable,
};
use crate::data::api_keys::{
    ApiKeyStoreError, CreateApiKeyParams, SaveApiKeyParams, create_api_key, create_api_key_right,
    delete_api_key, delete_api_key_right, delete_client_api_key_config, get_api_key,
    get_global_api_key_config, list_api_key_rights, list_api_keys, list_client_api_key_configs,
    save_api_key, set_global_api_key_config, update_api_key_right, upsert_client_api_key_config,
};
use crate::data::clients::{
    AthenaClientRecord, SaveAthenaClientParams, delete_athena_client, get_athena_client_by_name,
    get_client_statistics, list_athena_clients, list_client_statistics,
    list_client_table_statistics, refresh_client_statistics, set_client_frozen_state,
    upsert_athena_client,
};
use crate::drivers::postgresql::sqlx_driver::ClientConnectionTarget;

#[derive(Debug, Deserialize)]
struct CreateApiKeyRequest {
    name: String,
    #[serde(default)]
    description: Option<String>,
    #[serde(default)]
    client_name: Option<String>,
    #[serde(default)]
    expires_at: Option<String>,
    #[serde(default)]
    rights: Vec<String>,
}

#[derive(Debug, Deserialize)]
struct UpdateApiKeyRequest {
    #[serde(default)]
    name: Option<String>,
    #[serde(default)]
    description: Option<Option<String>>,
    #[serde(default)]
    client_name: Option<Option<String>>,
    #[serde(default)]
    expires_at: Option<Option<String>>,
    #[serde(default)]
    is_active: Option<bool>,
    #[serde(default)]
    rights: Option<Vec<String>>,
}

#[derive(Debug, Deserialize)]
struct SaveApiKeyRightRequest {
    name: String,
    #[serde(default)]
    description: Option<String>,
}

#[derive(Debug, Deserialize)]
struct SaveGlobalConfigRequest {
    enforce_api_keys: bool,
}

#[derive(Debug, Deserialize)]
struct SaveClientConfigRequest {
    enforce_api_keys: bool,
}

#[derive(Debug, Deserialize)]
struct SaveAthenaClientRequest {
    #[serde(default)]
    client_name: Option<String>,
    #[serde(default)]
    description: Option<String>,
    #[serde(default)]
    pg_uri: Option<String>,
    #[serde(default)]
    pg_uri_env_var: Option<String>,
    #[serde(default)]
    is_active: Option<bool>,
}

#[derive(Debug, Deserialize)]
struct FreezeAthenaClientRequest {
    is_frozen: bool,
}

fn auth_pool(state: &AppState) -> Result<PgPool, HttpResponse> {
    let Some(client_name) = state.gateway_auth_client_name.as_ref() else {
        return Err(service_unavailable(
            "Auth store unavailable",
            "No gateway auth client is configured.",
        ));
    };

    state.pg_registry.get_pool(client_name).ok_or_else(|| {
        service_unavailable(
            "Auth store unavailable",
            format!("Gateway auth client '{}' is not connected.", client_name),
        )
    })
}

fn client_catalog_pool(state: &AppState) -> Result<PgPool, HttpResponse> {
    let Some(client_name) = state.logging_client_name.as_ref() else {
        return Err(service_unavailable(
            "Client catalog unavailable",
            "No athena_logging client is configured.",
        ));
    };

    state.pg_registry.get_pool(client_name).ok_or_else(|| {
        service_unavailable(
            "Client catalog unavailable",
            format!("Logging client '{}' is not connected.", client_name),
        )
    })
}

fn normalize_client_name(client_name: &str) -> Result<String, HttpResponse> {
    let value = client_name.trim();
    if value.is_empty() {
        Err(bad_request(
            "Invalid client name",
            "The client name must not be empty.",
        ))
    } else {
        Ok(value.to_string())
    }
}

fn client_record_to_runtime_target(record: &AthenaClientRecord) -> ClientConnectionTarget {
    ClientConnectionTarget {
        client_name: record.client_name.clone(),
        source: record.source.clone(),
        description: record.description.clone(),
        pg_uri: record.pg_uri.clone(),
        pg_uri_env_var: record.pg_uri_env_var.clone(),
        config_uri_template: record.config_uri_template.clone(),
        is_active: record.is_active,
        is_frozen: record.is_frozen,
    }
}

async fn apply_client_record_to_runtime(
    state: &AppState,
    record: &AthenaClientRecord,
) -> Result<(), anyhow::Error> {
    let target = client_record_to_runtime_target(record);
    if !record.is_active || record.is_frozen {
        state.pg_registry.remember_client(target, false);
        state.pg_registry.mark_unavailable(&record.client_name);
        state.pg_registry.sync_connection_status();
        return Ok(());
    }

    if state.pg_registry.get_pool(&record.client_name).is_some() {
        state.pg_registry.remember_client(target, true);
        state.pg_registry.sync_connection_status();
        return Ok(());
    }

    state.pg_registry.upsert_client(target).await?;
    state.pg_registry.sync_connection_status();
    Ok(())
}

fn parse_optional_timestamp(
    value: Option<String>,
    field_name: &str,
) -> Result<Option<DateTime<Utc>>, HttpResponse> {
    match value {
        Some(value) => DateTime::parse_from_rfc3339(&value)
            .map(|timestamp| Some(timestamp.with_timezone(&Utc)))
            .map_err(|_| {
                bad_request(
                    "Invalid timestamp",
                    format!("'{}' must be an RFC3339 timestamp.", field_name),
                )
            }),
        None => Ok(None),
    }
}

fn apply_optional_timestamp_patch(
    patch_value: Option<Option<String>>,
    current: Option<DateTime<Utc>>,
    field_name: &str,
) -> Result<Option<DateTime<Utc>>, HttpResponse> {
    match patch_value {
        None => Ok(current),
        Some(None) => Ok(None),
        Some(Some(value)) => parse_optional_timestamp(Some(value), field_name),
    }
}

fn database_error_response(message: &str, err: sqlx::Error) -> HttpResponse {
    let unique_violation = err
        .as_database_error()
        .and_then(|db_err| db_err.code().map(|code| code == "23505"))
        .unwrap_or(false);

    if unique_violation {
        conflict(message, err.to_string())
    } else {
        internal_error(message, err.to_string())
    }
}

fn api_key_store_error_response(message: &str, err: ApiKeyStoreError) -> HttpResponse {
    match err {
        ApiKeyStoreError::MissingRights(missing) => bad_request(
            "Unknown API key rights",
            format!("These rights do not exist: {}", missing.join(", ")),
        ),
        ApiKeyStoreError::Database(err) => database_error_response(message, err),
    }
}

fn generate_public_id() -> String {
    Uuid::new_v4()
        .simple()
        .to_string()
        .chars()
        .take(16)
        .collect()
}

fn generate_secret() -> String {
    format!("{}{}", Uuid::new_v4().simple(), Uuid::new_v4().simple())
}

#[get("/admin/api-keys")]
async fn admin_list_api_keys(req: HttpRequest, app_state: Data<AppState>) -> impl Responder {
    if let Err(resp) = authorize_static_admin_key(&req) {
        return resp;
    }

    let pool = match auth_pool(app_state.get_ref()) {
        Ok(pool) => pool,
        Err(resp) => return resp,
    };

    match list_api_keys(&pool).await {
        Ok(keys) => api_success("Listed API keys", json!({ "api_keys": keys })),
        Err(err) => database_error_response("Failed to list API keys", err),
    }
}

#[post("/admin/api-keys")]
async fn admin_create_api_key(
    req: HttpRequest,
    body: Json<CreateApiKeyRequest>,
    app_state: Data<AppState>,
) -> impl Responder {
    if let Err(resp) = authorize_static_admin_key(&req) {
        return resp;
    }

    let pool = match auth_pool(app_state.get_ref()) {
        Ok(pool) => pool,
        Err(resp) => return resp,
    };

    let expires_at = match parse_optional_timestamp(body.expires_at.clone(), "expires_at") {
        Ok(value) => value,
        Err(resp) => return resp,
    };

    let public_id = generate_public_id();
    let secret = generate_secret();
    let key_salt = format!("{}{}", Uuid::new_v4().simple(), Uuid::new_v4().simple());
    let key_hash = digest(format!("{}:{}", key_salt, secret));

    let params = CreateApiKeyParams {
        public_id: public_id.clone(),
        name: body.name.clone(),
        description: body.description.clone(),
        client_name: body.client_name.clone(),
        key_salt,
        key_hash,
        expires_at,
        rights: body.rights.clone(),
    };

    match create_api_key(&pool, params).await {
        Ok(record) => api_created(
            "Created API key",
            json!({
                "api_key": format!("ath_{}.{}", public_id, secret),
                "record": record,
            }),
        ),
        Err(err) => api_key_store_error_response("Failed to create API key", err),
    }
}

#[patch("/admin/api-keys/{id}")]
async fn admin_update_api_key(
    req: HttpRequest,
    path: Path<String>,
    body: Json<UpdateApiKeyRequest>,
    app_state: Data<AppState>,
) -> impl Responder {
    if let Err(resp) = authorize_static_admin_key(&req) {
        return resp;
    }

    let pool = match auth_pool(app_state.get_ref()) {
        Ok(pool) => pool,
        Err(resp) => return resp,
    };

    let id = match Uuid::parse_str(&path.into_inner()) {
        Ok(value) => value,
        Err(_) => {
            return bad_request("Invalid API key id", "The API key id must be a UUID.");
        }
    };

    let Some(existing) = (match get_api_key(&pool, id).await {
        Ok(record) => record,
        Err(err) => return database_error_response("Failed to load API key", err),
    }) else {
        return not_found(
            "API key not found",
            format!("No API key exists for '{}'.", id),
        );
    };

    let expires_at = match apply_optional_timestamp_patch(
        body.expires_at.clone(),
        existing.expires_at,
        "expires_at",
    ) {
        Ok(value) => value,
        Err(resp) => return resp,
    };

    let params = SaveApiKeyParams {
        id,
        name: body.name.clone().unwrap_or(existing.name),
        description: body.description.clone().unwrap_or(existing.description),
        client_name: body.client_name.clone().unwrap_or(existing.client_name),
        expires_at,
        is_active: body.is_active.unwrap_or(existing.is_active),
        rights: body.rights.clone().unwrap_or(existing.rights),
    };

    match save_api_key(&pool, params).await {
        Ok(Some(record)) => api_success("Updated API key", json!({ "record": record })),
        Ok(None) => not_found(
            "API key not found",
            format!("No API key exists for '{}'.", id),
        ),
        Err(err) => api_key_store_error_response("Failed to update API key", err),
    }
}

#[delete("/admin/api-keys/{id}")]
async fn admin_delete_api_key(
    req: HttpRequest,
    path: Path<String>,
    app_state: Data<AppState>,
) -> impl Responder {
    if let Err(resp) = authorize_static_admin_key(&req) {
        return resp;
    }

    let pool = match auth_pool(app_state.get_ref()) {
        Ok(pool) => pool,
        Err(resp) => return resp,
    };

    let id = match Uuid::parse_str(&path.into_inner()) {
        Ok(value) => value,
        Err(_) => {
            return bad_request("Invalid API key id", "The API key id must be a UUID.");
        }
    };

    match delete_api_key(&pool, id).await {
        Ok(true) => api_success("Deleted API key", json!({ "id": id })),
        Ok(false) => not_found(
            "API key not found",
            format!("No API key exists for '{}'.", id),
        ),
        Err(err) => database_error_response("Failed to delete API key", err),
    }
}

#[get("/admin/api-key-rights")]
async fn admin_list_api_key_rights(req: HttpRequest, app_state: Data<AppState>) -> impl Responder {
    if let Err(resp) = authorize_static_admin_key(&req) {
        return resp;
    }

    let pool = match auth_pool(app_state.get_ref()) {
        Ok(pool) => pool,
        Err(resp) => return resp,
    };

    match list_api_key_rights(&pool).await {
        Ok(rights) => api_success("Listed API key rights", json!({ "rights": rights })),
        Err(err) => database_error_response("Failed to list API key rights", err),
    }
}

#[post("/admin/api-key-rights")]
async fn admin_create_api_key_right(
    req: HttpRequest,
    body: Json<SaveApiKeyRightRequest>,
    app_state: Data<AppState>,
) -> impl Responder {
    if let Err(resp) = authorize_static_admin_key(&req) {
        return resp;
    }

    let pool = match auth_pool(app_state.get_ref()) {
        Ok(pool) => pool,
        Err(resp) => return resp,
    };

    match create_api_key_right(&pool, &body.name, body.description.as_deref()).await {
        Ok(record) => api_created("Created API key right", json!({ "right": record })),
        Err(err) => database_error_response("Failed to create API key right", err),
    }
}

#[patch("/admin/api-key-rights/{id}")]
async fn admin_update_api_key_right(
    req: HttpRequest,
    path: Path<String>,
    body: Json<SaveApiKeyRightRequest>,
    app_state: Data<AppState>,
) -> impl Responder {
    if let Err(resp) = authorize_static_admin_key(&req) {
        return resp;
    }

    let pool = match auth_pool(app_state.get_ref()) {
        Ok(pool) => pool,
        Err(resp) => return resp,
    };

    let id = match Uuid::parse_str(&path.into_inner()) {
        Ok(value) => value,
        Err(_) => {
            return bad_request("Invalid API key right id", "The right id must be a UUID.");
        }
    };

    match update_api_key_right(&pool, id, &body.name, body.description.as_deref()).await {
        Ok(Some(record)) => api_success("Updated API key right", json!({ "right": record })),
        Ok(None) => not_found(
            "API key right not found",
            format!("No right exists for '{}'.", id),
        ),
        Err(err) => database_error_response("Failed to update API key right", err),
    }
}

#[delete("/admin/api-key-rights/{id}")]
async fn admin_delete_api_key_right(
    req: HttpRequest,
    path: Path<String>,
    app_state: Data<AppState>,
) -> impl Responder {
    if let Err(resp) = authorize_static_admin_key(&req) {
        return resp;
    }

    let pool = match auth_pool(app_state.get_ref()) {
        Ok(pool) => pool,
        Err(resp) => return resp,
    };

    let id = match Uuid::parse_str(&path.into_inner()) {
        Ok(value) => value,
        Err(_) => {
            return bad_request("Invalid API key right id", "The right id must be a UUID.");
        }
    };

    match delete_api_key_right(&pool, id).await {
        Ok(true) => api_success("Deleted API key right", json!({ "id": id })),
        Ok(false) => not_found(
            "API key right not found",
            format!("No right exists for '{}'.", id),
        ),
        Err(err) => database_error_response("Failed to delete API key right", err),
    }
}

#[get("/admin/api-key-config")]
async fn admin_get_api_key_config(req: HttpRequest, app_state: Data<AppState>) -> impl Responder {
    if let Err(resp) = authorize_static_admin_key(&req) {
        return resp;
    }

    let pool = match auth_pool(app_state.get_ref()) {
        Ok(pool) => pool,
        Err(resp) => return resp,
    };

    let global = match get_global_api_key_config(&pool).await {
        Ok(value) => value,
        Err(err) => return database_error_response("Failed to load API key config", err),
    };
    let clients = match list_client_api_key_configs(&pool).await {
        Ok(value) => value,
        Err(err) => return database_error_response("Failed to load API key client config", err),
    };

    api_success(
        "Loaded API key config",
        json!({
            "global": global,
            "clients": clients,
        }),
    )
}

#[put("/admin/api-key-config")]
async fn admin_set_api_key_config(
    req: HttpRequest,
    body: Json<SaveGlobalConfigRequest>,
    app_state: Data<AppState>,
) -> impl Responder {
    if let Err(resp) = authorize_static_admin_key(&req) {
        return resp;
    }

    let pool = match auth_pool(app_state.get_ref()) {
        Ok(pool) => pool,
        Err(resp) => return resp,
    };

    match set_global_api_key_config(&pool, body.enforce_api_keys).await {
        Ok(global) => api_success("Updated API key config", json!({ "global": global })),
        Err(err) => database_error_response("Failed to update API key config", err),
    }
}

#[get("/admin/api-key-clients")]
async fn admin_list_api_key_clients(req: HttpRequest, app_state: Data<AppState>) -> impl Responder {
    if let Err(resp) = authorize_static_admin_key(&req) {
        return resp;
    }

    let pool = match auth_pool(app_state.get_ref()) {
        Ok(pool) => pool,
        Err(resp) => return resp,
    };

    match list_client_api_key_configs(&pool).await {
        Ok(clients) => api_success(
            "Listed API key client config",
            json!({ "clients": clients }),
        ),
        Err(err) => database_error_response("Failed to list API key client config", err),
    }
}

#[put("/admin/api-key-clients/{client_name}")]
async fn admin_upsert_api_key_client(
    req: HttpRequest,
    path: Path<String>,
    body: Json<SaveClientConfigRequest>,
    app_state: Data<AppState>,
) -> impl Responder {
    if let Err(resp) = authorize_static_admin_key(&req) {
        return resp;
    }

    let pool: sqlx::Pool<sqlx::Postgres> = match auth_pool(app_state.get_ref()) {
        Ok(pool) => pool,
        Err(resp) => return resp,
    };

    let client_name: String = path.into_inner();
    if client_name.trim().is_empty() {
        return bad_request("Invalid client name", "The client name must not be empty.");
    }

    match upsert_client_api_key_config(&pool, &client_name, body.enforce_api_keys).await {
        Ok(record) => api_success("Updated API key client config", json!({ "client": record })),
        Err(err) => database_error_response("Failed to update API key client config", err),
    }
}

#[delete("/admin/api-key-clients/{client_name}")]
async fn admin_delete_api_key_client(
    req: HttpRequest,
    path: Path<String>,
    app_state: Data<AppState>,
) -> impl Responder {
    if let Err(resp) = authorize_static_admin_key(&req) {
        return resp;
    }

    let pool: sqlx::Pool<sqlx::Postgres> = match auth_pool(app_state.get_ref()) {
        Ok(pool) => pool,
        Err(resp) => return resp,
    };

    let client_name = path.into_inner();
    match delete_client_api_key_config(&pool, &client_name).await {
        Ok(true) => api_success(
            "Deleted API key client config",
            json!({ "client_name": client_name }),
        ),
        Ok(false) => not_found(
            "API key client config not found",
            format!("No client config exists for '{}'.", client_name),
        ),
        Err(err) => database_error_response("Failed to delete API key client config", err),
    }
}

#[get("/admin/clients")]
async fn admin_list_clients(req: HttpRequest, app_state: Data<AppState>) -> impl Responder {
    if let Err(resp) = authorize_static_admin_key(&req) {
        return resp;
    }

    let runtime_clients = app_state.pg_registry.list_registered_clients();
    let pool = match client_catalog_pool(app_state.get_ref()) {
        Ok(pool) => pool,
        Err(_) => {
            return api_success(
                "Listed runtime Athena clients",
                json!({ "clients": runtime_clients }),
            );
        }
    };

    match list_athena_clients(&pool).await {
        Ok(clients) => api_success(
            "Listed Athena clients",
            json!({
                "clients": clients,
                "runtime": runtime_clients,
            }),
        ),
        Err(err) => database_error_response("Failed to list Athena clients", err),
    }
}

#[post("/admin/clients")]
async fn admin_create_client(
    req: HttpRequest,
    body: Json<SaveAthenaClientRequest>,
    app_state: Data<AppState>,
) -> impl Responder {
    if let Err(resp) = authorize_static_admin_key(&req) {
        return resp;
    }

    let pool: sqlx::Pool<sqlx::Postgres> = match client_catalog_pool(app_state.get_ref()) {
        Ok(pool) => pool,
        Err(resp) => return resp,
    };

    let client_name = match normalize_client_name(body.client_name.as_deref().unwrap_or("")) {
        Ok(value) => value,
        Err(_) => {
            return bad_request(
                "Missing client name",
                "Provide client_name in the request body when creating a client.",
            );
        }
    };

    if body.pg_uri.as_deref().unwrap_or("").trim().is_empty()
        && body
            .pg_uri_env_var
            .as_deref()
            .unwrap_or("")
            .trim()
            .is_empty()
    {
        return bad_request(
            "Missing connection details",
            "Provide either pg_uri or pg_uri_env_var for database-backed clients.",
        );
    }

    match get_athena_client_by_name(&pool, &client_name).await {
        Ok(Some(_)) => {
            return conflict(
                "Athena client already exists",
                format!("A client named '{}' already exists.", client_name),
            );
        }
        Ok(None) => {}
        Err(err) => return database_error_response("Failed to load Athena client", err),
    }

    let record = match upsert_athena_client(
        &pool,
        SaveAthenaClientParams {
            client_name,
            description: body.description.clone(),
            pg_uri: body.pg_uri.clone(),
            pg_uri_env_var: body.pg_uri_env_var.clone(),
            config_uri_template: None,
            source: "database".to_string(),
            is_active: body.is_active.unwrap_or(true),
            is_frozen: false,
            metadata: json!({ "managed_by": "admin_api" }),
        },
    )
    .await
    {
        Ok(record) => record,
        Err(err) => return database_error_response("Failed to create Athena client", err),
    };

    if let Err(err) = apply_client_record_to_runtime(app_state.get_ref(), &record).await {
        return internal_error("Failed to activate Athena client", err.to_string());
    }

    api_created("Created Athena client", json!({ "client": record }))
}

#[patch("/admin/clients/{client_name}")]
async fn admin_update_client(
    req: HttpRequest,
    path: Path<String>,
    body: Json<SaveAthenaClientRequest>,
    app_state: Data<AppState>,
) -> impl Responder {
    if let Err(resp) = authorize_static_admin_key(&req) {
        return resp;
    }

    let pool = match client_catalog_pool(app_state.get_ref()) {
        Ok(pool) => pool,
        Err(resp) => return resp,
    };

    let client_name = match normalize_client_name(&path.into_inner()) {
        Ok(value) => value,
        Err(resp) => return resp,
    };

    let existing = match get_athena_client_by_name(&pool, &client_name).await {
        Ok(Some(record)) => record,
        Ok(None) => {
            return not_found(
                "Athena client not found",
                format!("No client exists for '{}'.", client_name),
            );
        }
        Err(err) => return database_error_response("Failed to load Athena client", err),
    };

    let record = match upsert_athena_client(
        &pool,
        SaveAthenaClientParams {
            client_name: client_name.clone(),
            description: body.description.clone().or(existing.description.clone()),
            pg_uri: body.pg_uri.clone().or(existing.pg_uri.clone()),
            pg_uri_env_var: body
                .pg_uri_env_var
                .clone()
                .or(existing.pg_uri_env_var.clone()),
            config_uri_template: existing.config_uri_template.clone(),
            source: existing.source.clone(),
            is_active: body.is_active.unwrap_or(existing.is_active),
            is_frozen: existing.is_frozen,
            metadata: existing.metadata.clone(),
        },
    )
    .await
    {
        Ok(record) => record,
        Err(err) => return database_error_response("Failed to update Athena client", err),
    };

    if let Err(err) = apply_client_record_to_runtime(app_state.get_ref(), &record).await {
        return internal_error("Failed to refresh Athena client", err.to_string());
    }

    api_success("Updated Athena client", json!({ "client": record }))
}

#[put("/admin/clients/{client_name}/freeze")]
async fn admin_freeze_client(
    req: HttpRequest,
    path: Path<String>,
    body: Json<FreezeAthenaClientRequest>,
    app_state: Data<AppState>,
) -> impl Responder {
    if let Err(resp) = authorize_static_admin_key(&req) {
        return resp;
    }

    let pool = match client_catalog_pool(app_state.get_ref()) {
        Ok(pool) => pool,
        Err(resp) => return resp,
    };

    let client_name = match normalize_client_name(&path.into_inner()) {
        Ok(value) => value,
        Err(resp) => return resp,
    };

    let record = match set_client_frozen_state(&pool, &client_name, body.is_frozen).await {
        Ok(Some(record)) => record,
        Ok(None) => {
            return not_found(
                "Athena client not found",
                format!("No client exists for '{}'.", client_name),
            );
        }
        Err(err) => return database_error_response("Failed to update Athena client", err),
    };

    if let Err(err) = apply_client_record_to_runtime(app_state.get_ref(), &record).await {
        return internal_error("Failed to refresh Athena client", err.to_string());
    }

    api_success(
        "Updated Athena client freeze state",
        json!({ "client": record }),
    )
}

#[delete("/admin/clients/{client_name}")]
async fn admin_delete_client(
    req: HttpRequest,
    path: Path<String>,
    app_state: Data<AppState>,
) -> impl Responder {
    if let Err(resp) = authorize_static_admin_key(&req) {
        return resp;
    }

    let pool = match client_catalog_pool(app_state.get_ref()) {
        Ok(pool) => pool,
        Err(resp) => return resp,
    };

    let client_name = match normalize_client_name(&path.into_inner()) {
        Ok(value) => value,
        Err(resp) => return resp,
    };

    match delete_athena_client(&pool, &client_name).await {
        Ok(Some(record)) => {
            app_state.pg_registry.remove_client(&record.client_name);
            api_success("Deleted Athena client", json!({ "client": record }))
        }
        Ok(None) => not_found(
            "Athena client not found",
            format!("No client exists for '{}'.", client_name),
        ),
        Err(err) => database_error_response("Failed to delete Athena client", err),
    }
}

#[get("/admin/clients/statistics")]
async fn admin_list_client_statistics(
    req: HttpRequest,
    app_state: Data<AppState>,
) -> impl Responder {
    if let Err(resp) = authorize_static_admin_key(&req) {
        return resp;
    }

    let pool = match client_catalog_pool(app_state.get_ref()) {
        Ok(pool) => pool,
        Err(resp) => return resp,
    };

    match list_client_statistics(&pool).await {
        Ok(statistics) => api_success(
            "Listed client statistics",
            json!({ "statistics": statistics }),
        ),
        Err(err) => database_error_response("Failed to list client statistics", err),
    }
}

#[post("/admin/clients/statistics/refresh")]
async fn admin_refresh_client_statistics(
    req: HttpRequest,
    app_state: Data<AppState>,
) -> impl Responder {
    if let Err(resp) = authorize_static_admin_key(&req) {
        return resp;
    }

    let pool = match client_catalog_pool(app_state.get_ref()) {
        Ok(pool) => pool,
        Err(resp) => return resp,
    };

    if let Err(err) = refresh_client_statistics(&pool).await {
        return database_error_response("Failed to refresh client statistics", err);
    }

    match list_client_statistics(&pool).await {
        Ok(statistics) => api_success(
            "Refreshed client statistics",
            json!({ "statistics": statistics }),
        ),
        Err(err) => database_error_response("Failed to list client statistics", err),
    }
}

#[get("/admin/clients/{client_name}/statistics")]
async fn admin_get_client_statistics(
    req: HttpRequest,
    path: Path<String>,
    app_state: Data<AppState>,
) -> impl Responder {
    if let Err(resp) = authorize_static_admin_key(&req) {
        return resp;
    }

    let pool = match client_catalog_pool(app_state.get_ref()) {
        Ok(pool) => pool,
        Err(resp) => return resp,
    };

    let client_name = match normalize_client_name(&path.into_inner()) {
        Ok(value) => value,
        Err(resp) => return resp,
    };

    let statistics = match get_client_statistics(&pool, &client_name).await {
        Ok(Some(record)) => record,
        Ok(None) => {
            return not_found(
                "Client statistics not found",
                format!("No statistics exist for '{}'.", client_name),
            );
        }
        Err(err) => return database_error_response("Failed to load client statistics", err),
    };

    match list_client_table_statistics(&pool, &client_name).await {
        Ok(table_statistics) => api_success(
            "Loaded client statistics",
            json!({
                "statistics": statistics,
                "tables": table_statistics,
            }),
        ),
        Err(err) => database_error_response("Failed to load client table statistics", err),
    }
}

pub fn services(cfg: &mut web::ServiceConfig) {
    cfg.service(admin_list_api_keys)
        .service(admin_create_api_key)
        .service(admin_update_api_key)
        .service(admin_delete_api_key)
        .service(admin_list_api_key_rights)
        .service(admin_create_api_key_right)
        .service(admin_update_api_key_right)
        .service(admin_delete_api_key_right)
        .service(admin_get_api_key_config)
        .service(admin_set_api_key_config)
        .service(admin_list_api_key_clients)
        .service(admin_upsert_api_key_client)
        .service(admin_delete_api_key_client)
        .service(admin_list_clients)
        .service(admin_create_client)
        .service(admin_update_client)
        .service(admin_freeze_client)
        .service(admin_delete_client)
        .service(admin_list_client_statistics)
        .service(admin_refresh_client_statistics)
        .service(admin_get_client_statistics);
}
