otelite_api/api/

logs.rs

use super::validate_time_range;
use crate::server::{AppState, QueryCache};
use axum::{
    extract::{Path, Query, State},
    http::StatusCode,
    response::IntoResponse,
    Json,
};
use otelite_core::api::{ErrorResponse, LogEntry, LogsResponse};
use otelite_core::query::{Operator, QueryPredicate, QueryValue};
use otelite_core::storage::QueryParams;
use otelite_core::telemetry::LogRecord;
use serde::{Deserialize, Serialize};

/// Query parameters for log listing
#[derive(Debug, Deserialize, Serialize, utoipa::IntoParams)]
pub struct LogsQuery {
    /// Filter by severity level (e.g., "ERROR", "WARN", "INFO")
    #[serde(default)]
    pub severity: Option<String>,

    /// Filter by resource attribute (e.g., "service.name=my-service")
    #[serde(default)]
    pub resource: Option<String>,

    /// Full-text search in log body
    #[serde(default)]
    pub search: Option<String>,

    /// Start time (Unix timestamp in nanoseconds)
    #[serde(default)]
    pub start_time: Option<i64>,

    /// End time (Unix timestamp in nanoseconds)
    #[serde(default)]
    pub end_time: Option<i64>,

    /// Filter by trace ID
    #[serde(default)]
    pub trace_id: Option<String>,

    /// Filter by session ID (session.id attribute)
    #[serde(default)]
    pub session_id: Option<String>,

    /// Filter by prompt ID (prompt.id attribute)
    #[serde(default)]
    pub prompt_id: Option<String>,

    /// Filter by gen_ai.conversation.id attribute
    #[serde(default)]
    pub conversation_id: Option<String>,

    /// Filter by model (matches gen_ai.request.model)
    #[serde(default)]
    pub model: Option<String>,

    /// JSON-encoded list of attribute filters: [{"key":"x","op":"=","value":"y"}, ...]
    /// Supported ops: "=" (Equal), "!=" (NotEqual)
    #[serde(default)]
    pub attrs: Option<String>,

    /// Structured query predicate; multiple predicates joined with `&&`,
    /// e.g. `body contains "timeout" && attributes.session.id = "abc"`.
    /// A malformed expression returns 400.
    #[serde(default)]
    pub query: Option<String>,

    /// Maximum number of results (default: 100, max: 1000)
    #[serde(default = "default_limit")]
    pub limit: usize,

    /// Offset for pagination
    #[serde(default)]
    pub offset: usize,
}

fn default_limit() -> usize {
    100
}

/// Handler for GET /api/logs
#[utoipa::path(
    get,
    path = "/api/logs",
    params(LogsQuery),
    responses(
        (status = 200, description = "List of logs matching query", body = LogsResponse),
        (status = 400, description = "Invalid time range", body = ErrorResponse),
        (status = 500, description = "Internal server error", body = ErrorResponse)
    ),
    tag = "logs"
)]
pub async fn list_logs(
    State(state): State<AppState>,
    Query(params): Query<LogsQuery>,
) -> Result<impl IntoResponse, (StatusCode, Json<ErrorResponse>)> {
    validate_time_range(params.start_time, params.end_time)?;

    // Check cache first
    let cache_key = QueryCache::make_key(&params);
    if let Some(cached) = state.cache.logs.get(&cache_key) {
        return Ok((
            StatusCode::OK,
            [("content-type", "application/json")],
            cached,
        )
            .into_response());
    }

    // Validate and cap limit
    let limit = params.limit.min(1000);

    // Build query parameters (treat empty strings as absent filters)
    let mut query = QueryParams {
        start_time: params.start_time,
        end_time: params.end_time,
        limit: Some(limit),
        search_text: params.search.filter(|s| !s.is_empty()),
        trace_id: params.trace_id.filter(|s| !s.is_empty()),
        ..Default::default()
    };

    // Parse severity filter if provided
    if let Some(severity_str) = params.severity.as_deref().filter(|s| !s.is_empty()) {
        query.min_severity = parse_severity(severity_str);
    }

    if let Some(ref sid) = params.session_id {
        if !sid.is_empty() {
            query.predicates.push(QueryPredicate {
                field: "session.id".to_string(),
                operator: Operator::Equal,
                value: QueryValue::String(sid.clone()),
            });
        }
    }

    if let Some(ref pid) = params.prompt_id {
        if !pid.is_empty() {
            query.predicates.push(QueryPredicate {
                field: "prompt.id".to_string(),
                operator: Operator::Equal,
                value: QueryValue::String(pid.clone()),
            });
        }
    }

    if let Some(ref cid) = params.conversation_id {
        if !cid.is_empty() {
            query.predicates.push(QueryPredicate {
                field: "gen_ai.conversation.id".to_string(),
                operator: Operator::Equal,
                value: QueryValue::String(cid.clone()),
            });
        }
    }

    if let Some(ref model) = params.model {
        if !model.is_empty() {
            query.predicates.push(QueryPredicate {
                field: "gen_ai.request.model".to_string(),
                operator: Operator::Equal,
                value: QueryValue::String(model.clone()),
            });
        }
    }

    if let Some(ref attrs_json) = params.attrs {
        if !attrs_json.is_empty() {
            #[derive(serde::Deserialize)]
            struct AttrFilter {
                key: String,
                op: String,
                value: Option<String>,
            }
            if let Ok(filters) = serde_json::from_str::<Vec<AttrFilter>>(attrs_json) {
                for f in filters {
                    let op = match f.op.as_str() {
                        "=" => Operator::Equal,
                        "!=" => Operator::NotEqual,
                        _ => continue,
                    };
                    if let Some(v) = f.value {
                        query.predicates.push(QueryPredicate {
                            field: f.key,
                            operator: op,
                            value: QueryValue::String(v),
                        });
                    }
                }
            }
        }
    }

    // Structured `query` predicate(s). Parse errors are a 400, never a
    // silently-ignored filter.
    if let Some(ref q) = params.query {
        if !q.is_empty() {
            query.predicates.extend(super::parse_query_param(q)?);
        }
    }

    // Query logs from storage
    let logs = state.storage.query_logs(&query).await.map_err(|e| {
        (
            StatusCode::INTERNAL_SERVER_ERROR,
            Json(ErrorResponse::storage_error(format!("query logs: {}", e))),
        )
    })?;

    // Filter by resource if specified (post-query filtering for now)
    let filtered_logs: Vec<LogRecord> =
        if let Some(resource_filter) = params.resource.as_deref().filter(|s| !s.is_empty()) {
            logs.into_iter()
                .filter(|log| matches_resource_filter(log, resource_filter))
                .collect()
        } else {
            logs
        };

    // Apply offset for pagination
    let total = filtered_logs.len();
    let paginated_logs: Vec<LogRecord> = filtered_logs
        .into_iter()
        .skip(params.offset)
        .take(limit)
        .collect();

    // Convert to API format, truncating bodies so list responses stay small.
    const MAX_LIST_BODY: usize = 512;
    let log_entries: Vec<LogEntry> = paginated_logs
        .into_iter()
        .map(|r| LogEntry::from(r).truncate_body(MAX_LIST_BODY))
        .collect();

    let response = LogsResponse {
        logs: log_entries,
        total,
        limit,
        offset: params.offset,
    };

    // Cache the response
    if let Ok(json) = serde_json::to_string(&response) {
        state.cache.logs.insert(cache_key, json.clone());
        Ok((StatusCode::OK, [("content-type", "application/json")], json).into_response())
    } else {
        Ok(Json(response).into_response())
    }
}

/// Handler for GET /api/logs/:timestamp
/// Note: Using timestamp as ID since LogRecord doesn't have a separate ID field
#[utoipa::path(
    get,
    path = "/api/logs/{timestamp}",
    params(
        ("timestamp" = i64, Path, description = "Log timestamp in nanoseconds")
    ),
    responses(
        (status = 200, description = "Log entry", body = LogEntry),
        (status = 404, description = "Log not found", body = ErrorResponse),
        (status = 500, description = "Internal server error", body = ErrorResponse)
    ),
    tag = "logs"
)]
pub async fn get_log(
    State(state): State<AppState>,
    Path(timestamp): Path<i64>,
) -> Result<impl IntoResponse, (StatusCode, Json<ErrorResponse>)> {
    // Query logs around this timestamp
    let query = QueryParams {
        start_time: Some(timestamp),
        end_time: Some(timestamp + 1),
        limit: Some(1),
        ..Default::default()
    };

    let logs = state.storage.query_logs(&query).await.map_err(|e| {
        (
            StatusCode::INTERNAL_SERVER_ERROR,
            Json(ErrorResponse::storage_error(format!(
                "query log by timestamp: {}",
                e
            ))),
        )
    })?;

    let log = logs.into_iter().next().ok_or_else(|| {
        (
            StatusCode::NOT_FOUND,
            Json(ErrorResponse::not_found(format!(
                "Log at timestamp {}",
                timestamp
            ))),
        )
    })?;

    Ok(Json(LogEntry::from(log)))
}

/// Export format for logs
#[derive(Debug, Deserialize, utoipa::IntoParams)]
pub struct ExportQuery {
    /// Export format: "json" or "csv"
    #[serde(default = "default_format")]
    pub format: String,

    /// Same filters as LogsQuery
    #[serde(flatten)]
    pub filters: LogsQuery,
}

fn default_format() -> String {
    "json".to_string()
}

/// Handler for GET /api/logs/export
#[utoipa::path(
    get,
    path = "/api/logs/export",
    params(ExportQuery),
    responses(
        (status = 200, description = "Exported logs in requested format"),
        (status = 400, description = "Invalid format parameter or time range", body = ErrorResponse),
        (status = 500, description = "Internal server error", body = ErrorResponse)
    ),
    tag = "logs"
)]
pub async fn export_logs(
    State(state): State<AppState>,
    Query(params): Query<ExportQuery>,
) -> Result<impl IntoResponse, (StatusCode, Json<ErrorResponse>)> {
    validate_time_range(params.filters.start_time, params.filters.end_time)?;

    // Build query parameters (no limit for export, but cap at 10000)
    let mut query = QueryParams {
        start_time: params.filters.start_time,
        end_time: params.filters.end_time,
        limit: Some(10000),
        search_text: params.filters.search.clone().filter(|s| !s.is_empty()),
        ..Default::default()
    };

    if let Some(severity_str) = params.filters.severity.as_deref().filter(|s| !s.is_empty()) {
        query.min_severity = parse_severity(severity_str);
    }

    let logs = state.storage.query_logs(&query).await.map_err(|e| {
        (
            StatusCode::INTERNAL_SERVER_ERROR,
            Json(ErrorResponse::storage_error(format!("export logs: {}", e))),
        )
    })?;

    // Filter by resource if specified (skip empty string = no filter)
    let filtered_logs: Vec<LogRecord> = if let Some(resource_filter) =
        params.filters.resource.as_deref().filter(|s| !s.is_empty())
    {
        logs.into_iter()
            .filter(|log| matches_resource_filter(log, resource_filter))
            .collect()
    } else {
        logs
    };

    match params.format.as_str() {
        "json" => {
            let log_entries: Vec<LogEntry> =
                filtered_logs.into_iter().map(LogEntry::from).collect();

            Ok((
                [
                    ("Content-Type", "application/json"),
                    ("Content-Disposition", "attachment; filename=\"logs.json\""),
                ],
                Json(log_entries),
            )
                .into_response())
        },
        "csv" => {
            // Simple CSV export
            let mut csv = String::from("timestamp,severity,body,trace_id,span_id\n");
            for log in filtered_logs {
                csv.push_str(&format!(
                    "{},{},{},{},{}\n",
                    log.timestamp,
                    log.severity.as_str(),
                    log.body.replace(',', ";").replace('\n', " "),
                    log.trace_id.unwrap_or_default(),
                    log.span_id.unwrap_or_default(),
                ));
            }

            Ok((
                [
                    ("Content-Type", "text/csv"),
                    ("Content-Disposition", "attachment; filename=\"logs.csv\""),
                ],
                csv,
            )
                .into_response())
        },
        _ => Err((
            StatusCode::BAD_REQUEST,
            Json(ErrorResponse::bad_request(
                "Invalid format parameter. Use 'json' or 'csv'",
            )),
        )),
    }
}

/// Parse severity string to SeverityLevel
fn parse_severity(s: &str) -> Option<otelite_core::telemetry::log::SeverityLevel> {
    use otelite_core::telemetry::log::SeverityLevel;
    match s.to_uppercase().as_str() {
        "TRACE" => Some(SeverityLevel::Trace),
        "DEBUG" => Some(SeverityLevel::Debug),
        "INFO" => Some(SeverityLevel::Info),
        "WARN" => Some(SeverityLevel::Warn),
        "ERROR" => Some(SeverityLevel::Error),
        "FATAL" => Some(SeverityLevel::Fatal),
        _ => None,
    }
}

/// Check if log matches resource filter (simple key=value matching)
fn matches_resource_filter(log: &LogRecord, filter: &str) -> bool {
    if let Some(resource) = &log.resource {
        if let Some((key, value)) = filter.split_once('=') {
            return resource.attributes.get(key).is_some_and(|v| v == value);
        }
    }
    false
}