athena-observability 3.18.0

//! Shared gateway log contracts, redaction helpers, and trace emitters.

use serde::Serialize;
use serde_json::{Map, Value};
use std::str::SplitN;
use tracing::{error, info};

/// Structured metadata returned to callers so downstream handlers can correlate their work.
#[derive(Debug, Clone)]
pub struct LoggedRequest {
    pub request_id: String,
    pub client_name: String,
    pub method: String,
    pub path: String,
    pub status_code: u16,
    pub time: i64,
}

#[derive(Clone, Debug, Default)]
pub struct RequestCompletionLogContext {
    pub status_code: Option<u16>,
    pub duration_ms: Option<u128>,
    pub cached: Option<bool>,
    pub cache_lookup_outcome: Option<String>,
    pub cache_source: Option<String>,
    pub operation: Option<String>,
    pub table_name: Option<String>,
}

#[derive(Clone, Debug, Default)]
pub struct RequestAuthLogContext {
    pub api_key_id: Option<String>,
    pub internal_api_key_id: Option<i64>,
    pub presented_api_key_public_id: Option<String>,
    pub presented_api_key_hash: Option<String>,
    pub presented_api_key_salt: Option<String>,
    pub api_key_authenticated: bool,
    pub api_key_authorized: bool,
    pub api_key_enforced: bool,
    pub api_key_auth_reason: Option<String>,
}

#[derive(Clone, Debug, Default)]
pub struct RouteRequestLogContext {
    pub request_id: Option<String>,
    pub source: Option<String>,
    pub outcome: Option<String>,
    pub method: Option<String>,
    pub path: Option<String>,
    pub host: Option<String>,
    pub wildcard_host_pattern: Option<String>,
    pub wildcard_prefix: Option<String>,
    pub route_key: Option<String>,
    pub public_op: Option<String>,
    pub resolved_client: Option<String>,
    pub status_code: Option<u16>,
    pub duration_ms: Option<u128>,
    pub error_message: Option<String>,
    pub metadata: Option<Value>,
}

#[derive(Serialize, Clone, Debug)]
pub struct RequestLogEntry {
    pub request_id: String,
    pub client: String,
    pub method: String,
    pub path: String,
    pub query_string: String,
    pub status_code: i32,
    pub ipv4: String,
    pub ipv4_address: String,
    pub user_agent: String,
    pub headers: Value,
    pub body: Value,
    pub user_id: String,
    pub company_id: String,
    pub organization_id: String,
    pub api_key_id: Option<String>,
    pub internal_api_key_id: Option<i64>,
    pub presented_api_key_public_id: Option<String>,
    pub presented_api_key_hash: Option<String>,
    pub presented_api_key_salt: Option<String>,
    pub api_key_authenticated: bool,
    pub api_key_authorized: bool,
    pub api_key_enforced: bool,
    pub api_key_auth_reason: Option<String>,
    pub host: String,
    pub athena_url: String,
    pub cached: bool,
    pub duration_ms: Option<i64>,
    pub cache_lookup_outcome: Option<String>,
    pub cache_source: Option<String>,
    pub operation: Option<String>,
    pub table_name: Option<String>,
    pub time: i64,
}

#[derive(Clone, Debug)]
pub struct RealtimeConnectionEntry {
    pub request_id: String,
    pub client_name: String,
    pub ipv4: String,
    pub athena_url: String,
    pub api_key_public_id: Option<String>,
    pub api_key_external_id: Option<String>,
    pub api_key_internal_id: Option<i64>,
    pub api_key_authenticated: bool,
    pub api_key_authorized: bool,
    pub method: String,
    pub path: String,
    pub user_agent: String,
    pub status_code: i32,
}

#[derive(Clone, Debug)]
pub struct RouteRequestLogEntry {
    pub request_id: String,
    pub source: String,
    pub outcome: String,
    pub method: Option<String>,
    pub path: Option<String>,
    pub host: Option<String>,
    pub wildcard_host_pattern: Option<String>,
    pub wildcard_prefix: Option<String>,
    pub route_key: Option<String>,
    pub public_op: Option<String>,
    pub resolved_client: Option<String>,
    pub status_code: Option<i32>,
    pub duration_ms: Option<i64>,
    pub error_message: Option<String>,
    pub metadata: Value,
    pub time: i64,
}

#[derive(Serialize, Clone, Debug)]
pub struct OperationLogEntry {
    pub request_id: String,
    pub operation: String,
    pub table_name: Option<String>,
    pub client: String,
    pub method: String,
    pub path: String,
    pub status_code: i32,
    pub duration_ms: i64,
    pub details: Value,
    pub time: i64,
    pub error: bool,
    pub message: Option<String>,
    pub cache_key: Option<String>,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub cache_key_hashed: Option<String>,
    pub cache_lookup_outcome: Option<String>,
    pub cache_source: Option<String>,
    pub cached: Option<bool>,
}

/// Returns `true` when a header/body/query key should be redacted.
pub fn is_sensitive_field_name(name: &str) -> bool {
    let normalized: String = name.to_ascii_lowercase().replace('-', "").replace('_', "");
    [
        "authorization",
        "apikey",
        "xapikey",
        "xathenakey",
        "cookie",
        "setcookie",
        "token",
        "secret",
        "password",
        "passwd",
        "jwt",
    ]
    .iter()
    .any(|needle| normalized.contains(needle))
}

/// Redacts sensitive query-string keys while leaving the rest untouched.
pub fn redact_query_string(query: &str) -> String {
    if query.is_empty() {
        return "NOT_FOUND".to_string();
    }

    query
        .split('&')
        .map(|part| {
            let mut split: SplitN<'_, char> = part.splitn(2, '=');
            let key: &str = split.next().unwrap_or_default();
            let value: Option<&str> = split.next();
            if is_sensitive_field_name(key) {
                if value.is_some() {
                    format!("{key}=[REDACTED]")
                } else {
                    format!("{key}=")
                }
            } else {
                part.to_string()
            }
        })
        .collect::<Vec<String>>()
        .join("&")
}

/// Recursively redacts sensitive fields inside JSON payloads.
pub fn redact_json_value(value: Value) -> Value {
    match value {
        Value::Object(map) => {
            let mut redacted: Map<String, Value> = Map::new();
            for (key, nested_value) in map {
                if is_sensitive_field_name(&key) {
                    redacted.insert(key, Value::String("[REDACTED]".to_string()));
                } else {
                    redacted.insert(key, redact_json_value(nested_value));
                }
            }
            Value::Object(redacted)
        }
        Value::Array(list) => Value::Array(list.into_iter().map(redact_json_value).collect()),
        other => other,
    }
}

/// Replaces the presence of a secret with a stable redaction marker.
pub fn redact_optional_secret(value: Option<String>) -> Option<String> {
    value.map(|_| "[REDACTED]".to_string())
}

fn is_gateway_trace_path(path: &str) -> bool {
    path == "/data"
        || path.starts_with("/gateway/")
        || path.starts_with("/rest/")
        || path.starts_with("/rpc/")
}

fn json_stringish_field(value: &Value, key: &str) -> Option<String> {
    let raw = value.get(key)?;
    match raw {
        Value::String(s) => Some(s.clone()),
        Value::Number(n) => Some(n.to_string()),
        Value::Bool(b) => Some(b.to_string()),
        _ => None,
    }
}

fn json_u64_field(value: &Value, key: &str) -> Option<u64> {
    let raw = value.get(key)?;
    match raw {
        Value::Number(n) => n
            .as_u64()
            .or_else(|| n.as_i64().and_then(|v| u64::try_from(v).ok())),
        Value::String(s) => s.parse::<u64>().ok(),
        _ => None,
    }
}

fn json_bool_field(value: &Value, key: &str) -> Option<bool> {
    let raw = value.get(key)?;
    match raw {
        Value::Bool(b) => Some(*b),
        Value::String(s) => match s.trim().to_ascii_lowercase().as_str() {
            "true" => Some(true),
            "false" => Some(false),
            _ => None,
        },
        _ => None,
    }
}

fn json_detail_keys(value: &Value) -> String {
    match value {
        Value::Object(map) => {
            let mut keys: Vec<&str> = map.keys().map(String::as_str).collect();
            keys.sort_unstable();
            keys.join(",")
        }
        _ => String::new(),
    }
}

fn gateway_trace_outcome(status_code: u16) -> &'static str {
    match status_code {
        200..=299 => "success",
        400..=499 => "client_error",
        500..=599 => "server_error",
        300..=399 => "redirect",
        _ => "other",
    }
}

/// Emits the structured tracing event used for gateway operation telemetry.
pub fn emit_gateway_operation_trace(
    request: &LoggedRequest,
    operation: &str,
    table_name: Option<&str>,
    duration_ms: u128,
    status_code: u16,
    details: &Value,
) {
    if !is_gateway_trace_path(&request.path) {
        return;
    }

    let outcome: &str = gateway_trace_outcome(status_code);
    let backend: String = json_stringish_field(details, "backend").unwrap_or_default();
    let error_code: String = json_stringish_field(details, "error_code").unwrap_or_default();
    let trace_id: String = json_stringish_field(details, "trace_id").unwrap_or_default();
    let cache_lookup_outcome: String =
        json_stringish_field(details, "cache_lookup_outcome").unwrap_or_default();
    let cache_source: String = json_stringish_field(details, "cache_source").unwrap_or_default();
    let resource_id: String = json_stringish_field(details, "resource_id").unwrap_or_default();
    let message: String = details
        .get("message")
        .and_then(Value::as_str)
        .or_else(|| details.get("error").and_then(Value::as_str))
        .unwrap_or_default()
        .to_string();
    let statement_count: u64 = json_u64_field(details, "statement_count").unwrap_or(0);
    let row_count: u64 = json_u64_field(details, "row_count")
        .or_else(|| json_u64_field(details, "updated_count"))
        .unwrap_or(0);
    let returned_row_count: u64 = json_u64_field(details, "returned_row_count").unwrap_or(0);
    let rows_affected: u64 = json_u64_field(details, "rows_affected").unwrap_or(0);
    let cached: bool = json_bool_field(details, "cached").unwrap_or(false);
    let detail_keys: String = json_detail_keys(details);
    let duration_ms_u64: u64 = duration_ms.min(u64::MAX as u128) as u64;

    if status_code >= 500 {
        error!(
            target: "athena_rs::gateway_trace",
            event = "gateway_operation",
            athena_version = env!("CARGO_PKG_VERSION"),
            request_id = %request.request_id,
            client = %request.client_name,
            method = %request.method,
            path = %request.path,
            operation = %operation,
            table_name = table_name.unwrap_or_default(),
            status_code = status_code,
            outcome = outcome,
            duration_ms = duration_ms_u64,
            backend = %backend,
            error_code = %error_code,
            trace_id = %trace_id,
            cache_lookup_outcome = %cache_lookup_outcome,
            cache_source = %cache_source,
            cached = cached,
            statement_count = statement_count,
            row_count = row_count,
            returned_row_count = returned_row_count,
            rows_affected = rows_affected,
            resource_id = %resource_id,
            detail_keys = %detail_keys,
            message = %message,
            "gateway operation completed"
        );
    } else {
        info!(
            target: "athena_rs::gateway_trace",
            event = "gateway_operation",
            athena_version = env!("CARGO_PKG_VERSION"),
            request_id = %request.request_id,
            client = %request.client_name,
            method = %request.method,
            path = %request.path,
            operation = %operation,
            table_name = table_name.unwrap_or_default(),
            status_code = status_code,
            outcome = outcome,
            duration_ms = duration_ms_u64,
            backend = %backend,
            error_code = %error_code,
            trace_id = %trace_id,
            cache_lookup_outcome = %cache_lookup_outcome,
            cache_source = %cache_source,
            cached = cached,
            statement_count = statement_count,
            row_count = row_count,
            returned_row_count = returned_row_count,
            rows_affected = rows_affected,
            resource_id = %resource_id,
            detail_keys = %detail_keys,
            message = %message,
            "gateway operation completed"
        );
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use serde_json::json;

    #[test]
    fn redact_query_string_hides_sensitive_values() {
        let redacted = redact_query_string("apikey=secret&limit=10");
        assert!(redacted.contains("apikey=[REDACTED]"));
        assert!(redacted.contains("limit=10"));
    }

    #[test]
    fn redact_json_value_hides_nested_secrets() {
        let value = json!({
            "password": "secret",
            "nested": {
                "token": "abc",
                "safe": 1
            }
        });
        let redacted = redact_json_value(value);
        assert_eq!(redacted["password"], "[REDACTED]");
        assert_eq!(redacted["nested"]["token"], "[REDACTED]");
        assert_eq!(redacted["nested"]["safe"], 1);
    }
}