use actix_web::{
HttpRequest, HttpResponse, Responder, delete, get, patch, post, put,
web::{self, Data, Json, Path, Query},
};
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, ClientOperationStatusFilter, SaveAthenaClientParams, delete_athena_client,
get_athena_client_by_name, get_client_statistics, list_athena_clients,
list_client_operation_drilldown, list_client_statistics, list_client_table_statistics,
refresh_client_statistics, set_client_frozen_state, upsert_athena_client,
};
use crate::data::vacuum_health::{
get_latest_vacuum_health_detail, list_latest_vacuum_health_summaries,
};
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,
}
#[derive(Debug, Deserialize)]
struct ClientStatisticsDrilldownQuery {
table_name: String,
operation: String,
#[serde(default)]
status: Option<String>,
#[serde(default)]
limit: Option<i64>,
#[serde(default)]
offset: Option<i64>,
}
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 normalize_required_query_value(field_name: &str, value: &str) -> Result<String, HttpResponse> {
let normalized = value.trim();
if normalized.is_empty() {
Err(bad_request(
"Invalid query parameter",
format!("'{}' must not be empty.", field_name),
))
} else {
Ok(normalized.to_string())
}
}
fn parse_client_statistics_status_filter(
status: Option<&str>,
) -> Result<ClientOperationStatusFilter, HttpResponse> {
match status.map(|value| value.trim().to_ascii_lowercase()) {
None => Ok(ClientOperationStatusFilter::All),
Some(value) if value.is_empty() || value == "all" => Ok(ClientOperationStatusFilter::All),
Some(value) if value == "errors" => Ok(ClientOperationStatusFilter::Errors),
Some(value) if value == "normal" => Ok(ClientOperationStatusFilter::Normal),
Some(_) => Err(bad_request(
"Invalid status filter",
"status must be one of: all, errors, normal",
)),
}
}
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),
}
}
#[get("/admin/clients/{client_name}/statistics/drilldown")]
async fn admin_get_client_statistics_drilldown(
req: HttpRequest,
path: Path<String>,
query: Query<ClientStatisticsDrilldownQuery>,
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 table_name = match normalize_required_query_value("table_name", &query.table_name) {
Ok(value) => value,
Err(resp) => return resp,
};
let operation = match normalize_required_query_value("operation", &query.operation) {
Ok(value) => value,
Err(resp) => return resp,
};
let status_filter = match parse_client_statistics_status_filter(query.status.as_deref()) {
Ok(value) => value,
Err(resp) => return resp,
};
let limit = query.limit.unwrap_or(100).clamp(1, 500);
let offset = query.offset.unwrap_or(0).max(0);
match list_client_operation_drilldown(
&pool,
&client_name,
&table_name,
&operation,
status_filter,
limit,
offset,
)
.await
{
Ok(rows) => api_success(
"Loaded client statistics drilldown",
json!({
"client_name": client_name,
"table_name": table_name,
"operation": operation,
"status": query.status.as_deref().map(str::trim).filter(|s| !s.is_empty()).unwrap_or("all"),
"limit": limit,
"offset": offset,
"rows": rows,
}),
),
Err(err) => database_error_response("Failed to load client statistics drilldown", err),
}
}
#[get("/admin/vacuum-health")]
async fn admin_list_vacuum_health(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_latest_vacuum_health_summaries(&pool).await {
Ok(snapshots) => api_success(
"Loaded vacuum health snapshots",
json!({
"snapshots": snapshots,
}),
),
Err(err) => database_error_response("Failed to load vacuum health snapshots", err),
}
}
#[get("/admin/vacuum-health/{client_name}")]
async fn admin_get_vacuum_health(
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 get_latest_vacuum_health_detail(&pool, &client_name).await {
Ok(Some((snapshot, tables))) => api_success(
"Loaded vacuum health detail",
json!({
"snapshot": snapshot,
"tables": tables,
}),
),
Ok(None) => not_found(
"Vacuum health not found",
format!("No vacuum health snapshots exist for '{}'.", client_name),
),
Err(err) => database_error_response("Failed to load vacuum health detail", 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)
.service(admin_get_client_statistics_drilldown)
.service(admin_list_vacuum_health)
.service(admin_get_vacuum_health);
}