solidb 1.0.2

//! HTTP handlers for Lua script management and execution

use axum::{
    extract::{Path, State},
    http::StatusCode,
    response::{IntoResponse, Json, Response},
};
use serde::{Deserialize, Serialize};
use serde_json::Value;
use std::collections::HashMap;

use super::auth::Claims;
use super::handlers::AppState;
use crate::error::DbError;
use crate::scripting::{Script, ScriptContext, ScriptEngine, ScriptUser, Service};
use crate::sync::{LogEntry, Operation};
use tracing::debug;

/// System collection for storing scripts
pub const SCRIPTS_COLLECTION: &str = "_scripts";

/// System collection for storing services
pub const SERVICES_COLLECTION: &str = "_services";

// ==================== Request/Response Types ====================

#[derive(Debug, Deserialize)]
pub struct CreateScriptRequest {
    /// Human-readable name for the script
    pub name: String,
    /// URL path pattern (e.g., "hello" or "users/:id")
    pub path: String,
    /// HTTP methods this script handles (e.g., ["GET", "POST"])
    pub methods: Vec<String>,
    /// The Lua source code
    pub code: String,
    /// Optional description
    pub description: Option<String>,
    /// Target collection (optional)
    pub collection: Option<String>,
    /// Service this script belongs to (defaults to "default")
    #[serde(default = "default_service")]
    pub service: String,
}

fn default_service() -> String {
    "default".to_string()
}

#[derive(Debug, Serialize)]
pub struct CreateScriptResponse {
    pub id: String,
    pub name: String,
    pub path: String,
    pub methods: Vec<String>,
    pub service: String,
    pub created_at: String,
}

#[derive(Debug, Serialize)]
pub struct ListScriptsResponse {
    pub scripts: Vec<ScriptSummary>,
}

#[derive(Debug, Serialize)]
pub struct ScriptSummary {
    pub id: String,
    pub name: String,
    pub path: String,
    pub methods: Vec<String>,
    pub description: Option<String>,
    pub database: String,
    pub service: String,
    pub collection: Option<String>,
    pub created_at: String,
    pub updated_at: String,
}

#[derive(Debug, Serialize)]
pub struct DeleteScriptResponse {
    pub deleted: bool,
}

#[derive(Debug, Serialize)]
pub struct ScriptStatsResponse {
    pub active_scripts: usize,
    pub active_ws: usize,
    pub total_scripts_executed: usize,
    pub total_ws_connections: usize,
}

// ==================== Script Management Handlers ====================

/// Create a new Lua script
pub async fn create_script_handler(
    State(state): State<AppState>,
    Path(db_name): Path<String>,
    Json(req): Json<CreateScriptRequest>,
) -> Result<Json<CreateScriptResponse>, DbError> {
    let db = state.storage.get_database(&db_name)?;

    // Ensure _scripts collection exists
    if db.get_collection(SCRIPTS_COLLECTION).is_err() {
        db.create_collection(SCRIPTS_COLLECTION.to_string(), None)?;
    }

    let collection = db.get_collection(SCRIPTS_COLLECTION)?;

    // Generate unique ID based on db/service/collection/path
    let path_key = sanitize_path_to_key(&req.path).ok_or_else(|| {
        DbError::BadRequest("Script path may not contain parent-dir traversal".to_string())
    })?;
    let id = if let Some(col) = &req.collection {
        format!("{}_{}_{}_{}", db_name, req.service, col, path_key)
    } else {
        format!("{}_{}_{}", db_name, req.service, path_key)
    };

    let now = chrono::Utc::now().to_rfc3339();

    // Check if script with same path already exists
    if collection.get(&id).is_ok() {
        return Err(DbError::BadRequest(format!(
            "Script with path '{}' already exists in this scope",
            req.path
        )));
    }

    let script = Script {
        key: id.clone(),
        name: req.name.clone(),
        methods: req.methods.clone(),
        path: req.path.clone(),
        database: db_name.clone(),
        service: req.service.clone(),
        collection: req.collection.clone(),
        code: req.code,
        description: req.description,
        created_at: now.clone(),
        updated_at: now.clone(),
    };

    let doc_value = serde_json::to_value(&script)
        .map_err(|e| DbError::InternalError(format!("Serialization error: {}", e)))?;

    collection.insert(doc_value.clone())?;

    tracing::info!(
        "Lua script '{}' created for path '{}' in db '{}'",
        req.name,
        req.path,
        db_name
    );

    // Update script index
    state.script_index.insert(script.clone());

    // Record write for replication
    if let Some(ref log) = state.replication_log {
        let entry = LogEntry {
            sequence: 0,
            node_id: "".to_string(), // Auto-filled
            database: db_name.clone(),
            collection: SCRIPTS_COLLECTION.to_string(),
            operation: Operation::Insert,
            key: id.clone(),
            data: serde_json::to_vec(&doc_value).ok(),
            timestamp: chrono::Utc::now().timestamp_millis() as u64,
            origin_sequence: None,
        };
        let _ = log.append(entry);
    }

    Ok(Json(CreateScriptResponse {
        id,
        name: req.name,
        path: req.path,
        methods: req.methods,
        service: req.service,
        created_at: now,
    }))
}

/// List scripts for a specific database
pub async fn list_scripts_handler(
    State(state): State<AppState>,
    Path(db_name): Path<String>,
) -> Result<Json<ListScriptsResponse>, DbError> {
    let db = state.storage.get_database(&db_name)?;

    // Return empty if collection doesn't exist
    let collection = match db.get_collection(SCRIPTS_COLLECTION) {
        Ok(c) => c,
        Err(DbError::CollectionNotFound(_)) => {
            return Ok(Json(ListScriptsResponse { scripts: vec![] }));
        }
        Err(e) => return Err(e),
    };

    let mut scripts = Vec::new();
    for doc in collection.scan(None) {
        let script: Script = serde_json::from_value(doc.to_value())
            .map_err(|_| DbError::InternalError("Corrupted script data".to_string()))?;

        // Filter by database
        if script.database == db_name {
            scripts.push(ScriptSummary {
                id: script.key,
                name: script.name,
                path: script.path,
                methods: script.methods,
                description: script.description,
                database: script.database,
                service: script.service,
                collection: script.collection,
                created_at: script.created_at,
                updated_at: script.updated_at,
            });
        }
    }

    Ok(Json(ListScriptsResponse { scripts }))
}

/// Get a specific script
pub async fn get_script_handler(
    State(state): State<AppState>,
    Path((db_name, script_id)): Path<(String, String)>,
) -> Result<Json<Script>, DbError> {
    let db = state.storage.get_database(&db_name)?;
    let collection = db.get_collection(SCRIPTS_COLLECTION)?;

    let doc = collection.get(&script_id)?;
    let script: Script = serde_json::from_value(doc.to_value())
        .map_err(|_| DbError::InternalError("Corrupted script data".to_string()))?;

    Ok(Json(script))
}

/// Update a script
pub async fn update_script_handler(
    State(state): State<AppState>,
    Path((db_name, script_id)): Path<(String, String)>,
    Json(req): Json<CreateScriptRequest>,
) -> Result<Json<Script>, DbError> {
    let db = state.storage.get_database(&db_name)?;
    let collection = db.get_collection(SCRIPTS_COLLECTION)?;

    // Get existing script to preserve sensitive fields
    let existing_doc = collection.get(&script_id)?;
    let existing: Script = serde_json::from_value(existing_doc.to_value())
        .map_err(|_| DbError::InternalError("Corrupted script data".to_string()))?;

    // We don't allow changing database, service, or collection effectively changing ID logic
    // So we persist existing database/service/collection
    let script = Script {
        key: script_id.clone(),
        name: req.name,
        methods: req.methods,
        path: req.path,
        database: existing.database,
        service: existing.service,
        collection: existing.collection,
        code: req.code,
        description: req.description,
        created_at: existing.created_at,
        updated_at: chrono::Utc::now().to_rfc3339(),
    };

    let doc_value = serde_json::to_value(&script)
        .map_err(|e| DbError::InternalError(format!("Serialization error: {}", e)))?;

    collection.update(&script_id, doc_value.clone())?;

    tracing::info!("Lua script '{}' updated", script_id);

    // Update script index (remove old, add new)
    state
        .script_index
        .remove(&script_id, &script.database, &script.service);
    state.script_index.insert(script.clone());

    // Invalidate bytecode cache for this script
    state.script_cache.invalidate(&script_id);

    // Record write for replication
    if let Some(ref log) = state.replication_log {
        let entry = LogEntry {
            sequence: 0,
            node_id: "".to_string(),
            database: db_name.clone(),
            collection: SCRIPTS_COLLECTION.to_string(),
            operation: Operation::Update,
            key: script_id.clone(),
            data: serde_json::to_vec(&doc_value).ok(),
            timestamp: chrono::Utc::now().timestamp_millis() as u64,
            origin_sequence: None,
        };
        let _ = log.append(entry);
    }

    Ok(Json(script))
}

/// Delete a script
pub async fn delete_script_handler(
    State(state): State<AppState>,
    Path((db_name, script_id)): Path<(String, String)>,
) -> Result<Json<DeleteScriptResponse>, DbError> {
    let db = state.storage.get_database(&db_name)?;
    let collection = db.get_collection(SCRIPTS_COLLECTION)?;

    // Get the script first to know its service for index removal
    let doc = collection.get(&script_id)?;
    let script: Script = serde_json::from_value(doc.to_value())
        .map_err(|_| DbError::InternalError("Corrupted script data".to_string()))?;

    collection.delete(&script_id)?;

    tracing::info!("Lua script '{}' deleted", script_id);

    // Remove from script index
    state
        .script_index
        .remove(&script_id, &db_name, &script.service);

    // Invalidate bytecode cache
    state.script_cache.invalidate(&script_id);

    // Record write for replication
    if let Some(ref log) = state.replication_log {
        let entry = LogEntry {
            sequence: 0,
            node_id: "".to_string(),
            database: db_name.clone(),
            collection: SCRIPTS_COLLECTION.to_string(),
            operation: Operation::Delete,
            key: script_id.clone(),
            data: None,
            timestamp: chrono::Utc::now().timestamp_millis() as u64,
            origin_sequence: None,
        };
        let _ = log.append(entry);
    }

    Ok(Json(DeleteScriptResponse { deleted: true }))
}

/// Get script runtime statistics
pub async fn get_script_stats_handler(
    State(state): State<AppState>,
) -> Result<Json<ScriptStatsResponse>, DbError> {
    use std::sync::atomic::Ordering;

    let stats = &state.script_stats;
    Ok(Json(ScriptStatsResponse {
        active_scripts: stats.active_scripts.load(Ordering::SeqCst),
        active_ws: stats.active_ws.load(Ordering::SeqCst),
        total_scripts_executed: stats.total_scripts_executed.load(Ordering::SeqCst),
        total_ws_connections: stats.total_ws_connections.load(Ordering::SeqCst),
    }))
}

// ==================== Helper Functions ====================

/// Convert a URL path to a valid document key.
/// Returns `None` when the path contains parent-dir traversal so callers
/// can reject with a 400 instead of producing a colliding empty/degenerate key.
fn sanitize_path_to_key(path: &str) -> Option<String> {
    let p = std::path::Path::new(path);
    if p.components()
        .any(|c| matches!(c, std::path::Component::ParentDir))
    {
        return None;
    }
    Some(
        path.replace(['/', ':', '*'], "_")
            .trim_matches('_')
            .to_string(),
    )
}

/// Check if a script path pattern matches the actual path
fn path_matches(pattern: &str, path: &str) -> bool {
    let pattern_parts: Vec<&str> = pattern.split('/').collect();
    let path_parts: Vec<&str> = path.split('/').collect();

    if pattern_parts.len() != path_parts.len() {
        return false;
    }

    for (p, actual) in pattern_parts.iter().zip(path_parts.iter()) {
        if p.starts_with(':') {
            // Parameter - matches anything
            continue;
        }
        if *p != *actual {
            return false;
        }
    }

    true
}

/// Extract parameters from the path based on the pattern
fn extract_path_params(pattern: &str, path: &str) -> HashMap<String, String> {
    let mut params = HashMap::new();
    let pattern_parts: Vec<&str> = pattern.split('/').collect();
    let path_parts: Vec<&str> = path.split('/').collect();

    if pattern_parts.len() != path_parts.len() {
        return params;
    }

    for (p, actual) in pattern_parts.iter().zip(path_parts.iter()) {
        if let Some(name) = p.strip_prefix(':') {
            params.insert(name.to_string(), actual.to_string());
        }
    }

    params
}

// ==================== REPL Types ====================

#[derive(Debug, Deserialize)]
pub struct ReplEvalRequest {
    /// Lua code to execute
    pub code: String,
    /// Optional session ID for state persistence
    pub session_id: Option<String>,
    /// Execution timeout in milliseconds (default 5000)
    #[serde(default = "default_timeout")]
    pub timeout_ms: u64,
}

fn default_timeout() -> u64 {
    5000
}

#[derive(Debug, Serialize)]
pub struct ReplEvalResponse {
    /// The returned value from the Lua code
    pub result: Value,
    /// Console output captured during execution
    pub output: Vec<String>,
    /// Error if execution failed
    #[serde(skip_serializing_if = "Option::is_none")]
    pub error: Option<ReplError>,
    /// Execution time in milliseconds
    pub execution_time_ms: f64,
    /// Session ID for subsequent calls
    pub session_id: String,
}

#[derive(Debug, Serialize)]
pub struct ReplError {
    /// Error message
    pub message: String,
    /// Line number where error occurred (if available)
    #[serde(skip_serializing_if = "Option::is_none")]
    pub line: Option<u32>,
    /// Column number where error occurred (if available)
    #[serde(skip_serializing_if = "Option::is_none")]
    pub column: Option<u32>,
}

// ==================== REPL Handler ====================

/// Evaluate Lua code in an interactive REPL session
pub async fn repl_eval_handler(
    State(state): State<AppState>,
    Path(db_name): Path<String>,
    axum::Extension(claims): axum::Extension<crate::server::auth::Claims>,
    Json(req): Json<ReplEvalRequest>,
) -> Result<Json<ReplEvalResponse>, DbError> {
    // Reject livequery tokens - they have limited privileges
    if claims.livequery == Some(true) {
        return Err(DbError::Forbidden(
            "REPL endpoint is not accessible with livequery tokens".to_string(),
        ));
    }

    if !crate::scripting::lua_runtime_enabled() {
        return Err(crate::scripting::lua_disabled_error());
    }

    // Require admin permission on the target database
    crate::server::authorization::AuthorizationService::check_permission(
        &claims,
        &state,
        crate::server::authorization::PermissionAction::Write,
        Some(&db_name),
    )
    .await?;

    use std::time::Instant;

    let start = Instant::now();

    // Verify database exists
    let _ = state.storage.get_database(&db_name)?;

    // Get or create session
    let mut session = state
        .repl_sessions
        .get_or_create(req.session_id.as_deref(), &db_name);

    // Get history BEFORE adding new code (so we don't replay current command)
    let history: Vec<String> = session.history.clone();

    // Now add the new code to history
    session.add_to_history(req.code.clone());

    // Create script engine
    let mut engine = ScriptEngine::new(state.storage.clone(), state.script_stats.clone());

    if let Some(sm) = &state.stream_manager {
        engine = engine.with_stream_manager(sm.clone());
    }

    // Execute with session variables and history for function replay
    let mut output_capture: Vec<String> = Vec::new();
    let result = engine
        .execute_repl(
            &req.code,
            &db_name,
            &session.variables,
            &history,
            &mut output_capture,
        )
        .await;

    let duration = start.elapsed();

    match result {
        Ok((value, updated_vars)) => {
            // Update session with new variables
            session.variables = updated_vars;
            state.repl_sessions.update(session.clone());

            Ok(Json(ReplEvalResponse {
                result: value,
                output: output_capture,
                error: None,
                execution_time_ms: duration.as_secs_f64() * 1000.0,
                session_id: session.id,
            }))
        }
        Err(e) => {
            // Parse error for line/column info
            let (message, line, column) = parse_lua_error(&e.to_string());

            Ok(Json(ReplEvalResponse {
                result: Value::Null,
                output: output_capture,
                error: Some(ReplError {
                    message,
                    line,
                    column,
                }),
                execution_time_ms: duration.as_secs_f64() * 1000.0,
                session_id: session.id,
            }))
        }
    }
}

// ==================== Service Management Types ====================

#[derive(Debug, Deserialize)]
pub struct CreateServiceRequest {
    /// Service identifier (e.g., "users", "auth")
    pub key: String,
    /// Human-readable name
    pub name: String,
    /// Optional description
    pub description: Option<String>,
    /// API version (e.g., "1.0.0")
    pub version: Option<String>,
    /// Whether this service is enabled
    #[serde(default = "default_enabled")]
    pub enabled: bool,
    /// Default auth requirement for scripts in this service
    #[serde(default = "default_require_auth")]
    pub require_auth: bool,
}

fn default_enabled() -> bool {
    true
}

fn default_require_auth() -> bool {
    true
}

#[derive(Debug, Serialize)]
pub struct CreateServiceResponse {
    pub key: String,
    pub name: String,
    pub database: String,
    pub created_at: String,
}

#[derive(Debug, Serialize)]
pub struct ListServicesResponse {
    pub services: Vec<ServiceSummary>,
}

#[derive(Debug, Serialize)]
pub struct ServiceSummary {
    pub key: String,
    pub name: String,
    pub description: Option<String>,
    pub version: Option<String>,
    pub database: String,
    pub enabled: bool,
    pub require_auth: bool,
    pub script_count: usize,
    pub created_at: String,
    pub updated_at: String,
}

#[derive(Debug, Deserialize)]
pub struct UpdateServiceRequest {
    /// Human-readable name
    pub name: Option<String>,
    /// Optional description
    pub description: Option<String>,
    /// API version (e.g., "1.0.0")
    pub version: Option<String>,
    /// Whether this service is enabled
    pub enabled: Option<bool>,
    /// Default auth requirement for scripts in this service
    pub require_auth: Option<bool>,
}

#[derive(Debug, Serialize)]
pub struct DeleteServiceResponse {
    pub deleted: bool,
    pub scripts_deleted: usize,
}

// ==================== Service Management Handlers ====================

/// Create a new service
pub async fn create_service_handler(
    State(state): State<AppState>,
    Path(db_name): Path<String>,
    Json(req): Json<CreateServiceRequest>,
) -> Result<Json<CreateServiceResponse>, DbError> {
    let db = state.storage.get_database(&db_name)?;

    // Ensure _services collection exists
    if db.get_collection(SERVICES_COLLECTION).is_err() {
        db.create_collection(SERVICES_COLLECTION.to_string(), None)?;
    }

    let collection = db.get_collection(SERVICES_COLLECTION)?;

    // Check if service already exists
    if collection.get(&req.key).is_ok() {
        return Err(DbError::BadRequest(format!(
            "Service '{}' already exists in database '{}'",
            req.key, db_name
        )));
    }

    let now = chrono::Utc::now().to_rfc3339();

    let service = Service {
        key: req.key.clone(),
        name: req.name.clone(),
        description: req.description,
        version: req.version,
        database: db_name.clone(),
        enabled: req.enabled,
        require_auth: req.require_auth,
        created_at: now.clone(),
        updated_at: now.clone(),
    };

    let doc_value = serde_json::to_value(&service)
        .map_err(|e| DbError::InternalError(format!("Serialization error: {}", e)))?;

    collection.insert(doc_value.clone())?;

    tracing::info!("Service '{}' created in database '{}'", req.key, db_name);
    state.service_cache.insert(&db_name, &req.key, service);

    // Record write for replication
    if let Some(ref log) = state.replication_log {
        let entry = LogEntry {
            sequence: 0,
            node_id: "".to_string(),
            database: db_name.clone(),
            collection: SERVICES_COLLECTION.to_string(),
            operation: Operation::Insert,
            key: req.key.clone(),
            data: serde_json::to_vec(&doc_value).ok(),
            timestamp: chrono::Utc::now().timestamp_millis() as u64,
            origin_sequence: None,
        };
        let _ = log.append(entry);
    }

    Ok(Json(CreateServiceResponse {
        key: req.key,
        name: req.name,
        database: db_name,
        created_at: now,
    }))
}

/// List all services for a database
pub async fn list_services_handler(
    State(state): State<AppState>,
    Path(db_name): Path<String>,
) -> Result<Json<ListServicesResponse>, DbError> {
    let db = state.storage.get_database(&db_name)?;

    // Return empty if collection doesn't exist
    let collection = match db.get_collection(SERVICES_COLLECTION) {
        Ok(c) => c,
        Err(DbError::CollectionNotFound(_)) => {
            return Ok(Json(ListServicesResponse { services: vec![] }));
        }
        Err(e) => return Err(e),
    };

    // Also get scripts collection for counting
    let scripts: Vec<Script> = match db.get_collection(SCRIPTS_COLLECTION) {
        Ok(scripts_col) => scripts_col
            .scan(None)
            .into_iter()
            .filter_map(|doc| serde_json::from_value::<Script>(doc.to_value()).ok())
            .filter(|s| s.database == db_name)
            .collect(),
        Err(_) => vec![],
    };

    let mut services = Vec::new();
    for doc in collection.scan(None) {
        let service: Service = serde_json::from_value(doc.to_value())
            .map_err(|_| DbError::InternalError("Corrupted service data".to_string()))?;

        if service.database == db_name {
            let script_count = scripts.iter().filter(|s| s.service == service.key).count();
            services.push(ServiceSummary {
                key: service.key,
                name: service.name,
                description: service.description,
                version: service.version,
                database: service.database,
                enabled: service.enabled,
                require_auth: service.require_auth,
                script_count,
                created_at: service.created_at,
                updated_at: service.updated_at,
            });
        }
    }

    Ok(Json(ListServicesResponse { services }))
}

/// Get a specific service
pub async fn get_service_handler(
    State(state): State<AppState>,
    Path((db_name, service_key)): Path<(String, String)>,
) -> Result<Json<Service>, DbError> {
    let db = state.storage.get_database(&db_name)?;
    let collection = db.get_collection(SERVICES_COLLECTION)?;

    let doc = collection.get(&service_key)?;
    let service: Service = serde_json::from_value(doc.to_value())
        .map_err(|_| DbError::InternalError("Corrupted service data".to_string()))?;

    Ok(Json(service))
}

/// Update a service
pub async fn update_service_handler(
    State(state): State<AppState>,
    Path((db_name, service_key)): Path<(String, String)>,
    Json(req): Json<UpdateServiceRequest>,
) -> Result<Json<Service>, DbError> {
    let db = state.storage.get_database(&db_name)?;
    let collection = db.get_collection(SERVICES_COLLECTION)?;

    // Get existing service
    let existing_doc = collection.get(&service_key)?;
    let existing: Service = serde_json::from_value(existing_doc.to_value())
        .map_err(|_| DbError::InternalError("Corrupted service data".to_string()))?;

    let service = Service {
        key: existing.key,
        name: req.name.unwrap_or(existing.name),
        description: req.description.or(existing.description),
        version: req.version.or(existing.version),
        database: existing.database,
        enabled: req.enabled.unwrap_or(existing.enabled),
        require_auth: req.require_auth.unwrap_or(existing.require_auth),
        created_at: existing.created_at,
        updated_at: chrono::Utc::now().to_rfc3339(),
    };

    let doc_value = serde_json::to_value(&service)
        .map_err(|e| DbError::InternalError(format!("Serialization error: {}", e)))?;

    collection.update(&service_key, doc_value.clone())?;

    tracing::info!(
        "Service '{}' updated in database '{}'",
        service_key,
        db_name
    );
    state
        .service_cache
        .insert(&db_name, &service_key, service.clone());

    // Record write for replication
    if let Some(ref log) = state.replication_log {
        let entry = LogEntry {
            sequence: 0,
            node_id: "".to_string(),
            database: db_name.clone(),
            collection: SERVICES_COLLECTION.to_string(),
            operation: Operation::Update,
            key: service_key.clone(),
            data: serde_json::to_vec(&doc_value).ok(),
            timestamp: chrono::Utc::now().timestamp_millis() as u64,
            origin_sequence: None,
        };
        let _ = log.append(entry);
    }

    Ok(Json(service))
}

/// Delete a service and all its scripts (cascade delete)
pub async fn delete_service_handler(
    State(state): State<AppState>,
    Path((db_name, service_key)): Path<(String, String)>,
) -> Result<Json<DeleteServiceResponse>, DbError> {
    let db = state.storage.get_database(&db_name)?;

    // First delete all scripts belonging to this service
    let mut scripts_deleted = 0;
    if let Ok(scripts_col) = db.get_collection(SCRIPTS_COLLECTION) {
        let scripts_to_delete: Vec<String> = scripts_col
            .scan(None)
            .into_iter()
            .filter_map(|doc| {
                serde_json::from_value::<Script>(doc.to_value())
                    .ok()
                    .filter(|s| s.database == db_name && s.service == service_key)
                    .map(|s| s.key)
            })
            .collect();

        for script_key in &scripts_to_delete {
            if scripts_col.delete(script_key).is_ok() {
                state
                    .script_index
                    .remove(script_key, &db_name, &service_key);
                state.script_cache.invalidate(script_key);
                scripts_deleted += 1;

                // Record delete for replication
                if let Some(ref log) = state.replication_log {
                    let entry = LogEntry {
                        sequence: 0,
                        node_id: "".to_string(),
                        database: db_name.clone(),
                        collection: SCRIPTS_COLLECTION.to_string(),
                        operation: Operation::Delete,
                        key: script_key.clone(),
                        data: None,
                        timestamp: chrono::Utc::now().timestamp_millis() as u64,
                        origin_sequence: None,
                    };
                    let _ = log.append(entry);
                }
            }
        }
    }

    // Now delete the service
    let collection = db.get_collection(SERVICES_COLLECTION)?;
    collection.delete(&service_key)?;

    tracing::info!(
        "Service '{}' deleted from database '{}' ({} scripts deleted)",
        service_key,
        db_name,
        scripts_deleted
    );
    state.service_cache.invalidate(&db_name, &service_key);

    // Record write for replication
    if let Some(ref log) = state.replication_log {
        let entry = LogEntry {
            sequence: 0,
            node_id: "".to_string(),
            database: db_name.clone(),
            collection: SERVICES_COLLECTION.to_string(),
            operation: Operation::Delete,
            key: service_key.clone(),
            data: None,
            timestamp: chrono::Utc::now().timestamp_millis() as u64,
            origin_sequence: None,
        };
        let _ = log.append(entry);
    }

    Ok(Json(DeleteServiceResponse {
        deleted: true,
        scripts_deleted,
    }))
}

/// Get OpenAPI spec for a specific service
pub async fn get_service_openapi_handler(
    State(state): State<AppState>,
    Path((db_name, service_key)): Path<(String, String)>,
) -> Result<Json<Value>, DbError> {
    let db = state.storage.get_database(&db_name)?;

    // Get service info
    let services_col = db.get_collection(SERVICES_COLLECTION)?;
    let service_doc = services_col.get(&service_key)?;
    let service: Service = serde_json::from_value(service_doc.to_value())
        .map_err(|_| DbError::InternalError("Corrupted service data".to_string()))?;

    // Get scripts for this service
    let scripts: Vec<Script> = match db.get_collection(SCRIPTS_COLLECTION) {
        Ok(scripts_col) => scripts_col
            .scan(None)
            .into_iter()
            .filter_map(|doc| serde_json::from_value::<Script>(doc.to_value()).ok())
            .filter(|s| s.database == db_name && s.service == service_key)
            .collect(),
        Err(_) => vec![],
    };

    // Build OpenAPI spec
    let mut paths = serde_json::Map::new();
    for script in scripts {
        let path_key = format!("/api/{}/{}/{}", db_name, service_key, script.path);
        let mut methods = serde_json::Map::new();

        for method in &script.methods {
            let method_lower = method.to_lowercase();
            if method_lower == "ws" {
                continue; // Skip WebSocket methods for OpenAPI
            }

            let mut operation = serde_json::Map::new();
            operation.insert("summary".to_string(), Value::String(script.name.clone()));
            if let Some(ref desc) = script.description {
                operation.insert("description".to_string(), Value::String(desc.clone()));
            }
            operation.insert(
                "operationId".to_string(),
                Value::String(format!("{}_{}", method_lower, script.key)),
            );

            // Add tags
            operation.insert(
                "tags".to_string(),
                Value::Array(vec![Value::String(service_key.clone())]),
            );

            // Default responses
            let responses = serde_json::json!({
                "200": {
                    "description": "Successful response"
                }
            });
            operation.insert("responses".to_string(), responses);

            methods.insert(method_lower, Value::Object(operation));
        }

        paths.insert(path_key, Value::Object(methods));
    }

    let openapi = serde_json::json!({
        "openapi": "3.0.0",
        "info": {
            "title": service.name,
            "description": service.description,
            "version": service.version.unwrap_or_else(|| "1.0.0".to_string())
        },
        "servers": [
            {
                "url": format!("/api/{}/{}", db_name, service_key),
                "description": "Service API endpoint"
            }
        ],
        "paths": paths
    });

    Ok(Json(openapi))
}

// ==================== Service Script Execution ====================

/// Execute a Lua script via service-based routing: /api/{db}/{service}/{path}
pub async fn execute_service_script_handler(
    State(state): State<AppState>,
    claims: Option<axum::Extension<Claims>>,
    ws_res: Result<
        axum::extract::ws::WebSocketUpgrade,
        axum::extract::ws::rejection::WebSocketUpgradeRejection,
    >,
    method: axum::http::Method,
    axum::extract::OriginalUri(uri): axum::extract::OriginalUri,
    headers: axum::http::HeaderMap,
    body: Option<Json<Value>>,
) -> Result<Response, DbError> {
    if !crate::scripting::lua_runtime_enabled() {
        return Err(crate::scripting::lua_disabled_error());
    }

    // Parse /api/{db}/{service}/{path}
    let uri_path = uri.path().to_string();
    let prefix = "/api/";
    let remaining = uri_path.strip_prefix(prefix).unwrap_or(&uri_path);

    // Split into db/service/path
    let parts: Vec<&str> = remaining.splitn(3, '/').collect();
    if parts.len() < 2 {
        return Err(DbError::BadRequest(
            "Invalid API path. Expected /api/{db}/{service}/{path}".to_string(),
        ));
    }

    let db_name = parts[0];
    let service_key = parts[1];
    let script_path = if parts.len() > 2 { parts[2] } else { "" };

    // Verify service exists and is enabled (cached)
    let service = if let Some(cached) = state.service_cache.get(db_name, service_key) {
        if !cached.enabled {
            return Err(DbError::BadRequest(format!(
                "Service '{}' is disabled",
                service_key
            )));
        }
        cached
    } else {
        let db = state.storage.get_database(db_name)?;
        let services_col = match db.get_collection(SERVICES_COLLECTION) {
            Ok(c) => c,
            Err(DbError::CollectionNotFound(_)) => {
                return Err(DbError::DocumentNotFound(format!(
                    "Service '{}' not found in database '{}'",
                    service_key, db_name
                )));
            }
            Err(e) => return Err(e),
        };

        match services_col.get(service_key) {
            Ok(doc) => {
                let s: Service = serde_json::from_value(doc.to_value())
                    .map_err(|_| DbError::InternalError("Corrupted service data".to_string()))?;
                if !s.enabled {
                    return Err(DbError::BadRequest(format!(
                        "Service '{}' is disabled",
                        service_key
                    )));
                }
                state.service_cache.insert(db_name, service_key, s.clone());
                s
            }
            Err(DbError::DocumentNotFound(_)) => {
                return Err(DbError::DocumentNotFound(format!(
                    "Service '{}' not found in database '{}'",
                    service_key, db_name
                )));
            }
            Err(e) => return Err(e),
        }
    };

    // Check service-level auth if required
    if service.require_auth && claims.is_none() {
        return Err(DbError::Unauthorized(
            "Authentication required for this service".to_string(),
        ));
    }

    let is_ws_upgrade = ws_res.is_ok();

    // Find matching script using the index (fast path)
    let script = match state
        .script_index
        .find(db_name, service_key, script_path, method.as_str())
    {
        Some(s) => {
            debug!(
                "Script found in index for {} {}/{}/{} in {}",
                method, db_name, service_key, script_path, db_name
            );
            s
        }
        None => {
            // Fallback to collection scan
            debug!(
                "Script not in index, falling back to scan for {} {}/{}/{} in {}",
                method, db_name, service_key, script_path, db_name
            );
            find_script_for_service_path(
                &state,
                db_name,
                service_key,
                script_path,
                method.as_str(),
                is_ws_upgrade,
            )?
        }
    };

    // Build context
    let query_params: HashMap<String, String> = uri
        .query()
        .map(|q| {
            url::form_urlencoded::parse(q.as_bytes())
                .into_owned()
                .collect()
        })
        .unwrap_or_default();

    let headers_map: HashMap<String, String> = headers
        .iter()
        .filter_map(|(k, v)| {
            v.to_str()
                .ok()
                .map(|v| (k.as_str().to_string(), v.to_string()))
        })
        .collect();

    // Build ScriptUser from claims if authenticated
    let user = match claims {
        Some(axum::Extension(c)) => ScriptUser {
            username: c.sub.clone(),
            roles: c.roles.clone().unwrap_or_default(),
            authenticated: true,
            scoped_databases: c.scoped_databases.clone(),
            exp: Some(c.exp as u64),
        },
        None => ScriptUser::anonymous(),
    };

    let context = ScriptContext {
        method: method.to_string(),
        path: script_path.to_string(),
        query_params,
        params: extract_path_params(&script.path, script_path),
        headers: headers_map,
        body: body.map(|b| b.0),
        is_websocket: ws_res.is_ok()
            && headers
                .get("upgrade")
                .and_then(|v| v.to_str().ok())
                .unwrap_or_default()
                .to_lowercase()
                == "websocket",
        user,
    };

    // Execute script with pooled Lua VM and bytecode cache
    let mut engine = ScriptEngine::new(state.storage.clone(), state.script_stats.clone())
        .with_script_cache(state.script_cache.clone());
    if let Some(pool) = &state.lua_pool {
        engine = engine.with_lua_pool(pool.clone());
    }

    if let Some(sm) = &state.stream_manager {
        engine = engine.with_stream_manager(sm.clone());
    }
    engine = engine.with_channel_manager(state.channel_manager.clone());

    // Handle WebSocket upgrade
    if context.is_websocket {
        if let Ok(ws) = ws_res {
            let db_name = db_name.to_string();
            return Ok(ws
                .on_upgrade(move |socket| async move {
                    if let Err(e) = engine.execute_ws(&script, &db_name, &context, socket).await {
                        tracing::error!("WebSocket script execution failed: {}", e);
                    }
                })
                .into_response());
        }
    }

    // Auto-select DB in Lua context using the path's db_name
    let result = engine.execute(&script, db_name, &context).await?;

    let status = StatusCode::from_u16(result.status).unwrap_or(StatusCode::OK);

    // Fast path: if script returned pre-serialized JSON, send it directly
    if let Some(raw) = result.raw_body {
        return Ok((
            status,
            [(axum::http::header::CONTENT_TYPE, "application/json")],
            raw,
        )
            .into_response());
    }

    Ok((status, Json(result.body)).into_response())
}

/// Find a script that matches the given service path and method
fn find_script_for_service_path(
    state: &AppState,
    db_name: &str,
    service_key: &str,
    path: &str,
    method: &str,
    is_ws_upgrade: bool,
) -> Result<Script, DbError> {
    let db = state.storage.get_database(db_name)?;
    let collection = db.get_collection(SCRIPTS_COLLECTION)?;

    for doc in collection.scan(None) {
        let script: Script = match serde_json::from_value(doc.to_value()) {
            Ok(s) => s,
            Err(_) => continue,
        };

        // Check database and service
        if script.database != db_name || script.service != service_key {
            continue;
        }

        // Check if method matches
        if !script.methods.iter().any(|m| {
            m.eq_ignore_ascii_case(method) || (is_ws_upgrade && m.eq_ignore_ascii_case("WS"))
        }) {
            continue;
        }

        // Check if path matches
        if path_matches(&script.path, path) {
            return Ok(script);
        }
    }

    Err(DbError::DocumentNotFound(format!(
        "No script found for {} {}/{}/{} in {}",
        method, db_name, service_key, path, db_name
    )))
}

/// Parse a Lua error message to extract line/column information
fn parse_lua_error(error: &str) -> (String, Option<u32>, Option<u32>) {
    // Lua errors often look like: "[string \"...\"]:3: error message"
    // or "runtime error: [string \"...\"]:5:12: message"

    let re_line = regex::Regex::new(r"\[string [^\]]+\]:(\d+):(?:(\d+):)?\s*(.*)").ok();

    if let Some(re) = re_line {
        if let Some(caps) = re.captures(error) {
            let line = caps.get(1).and_then(|m| m.as_str().parse().ok());
            let column = caps.get(2).and_then(|m| m.as_str().parse().ok());
            let message = caps
                .get(3)
                .map(|m| m.as_str().to_string())
                .unwrap_or_else(|| error.to_string());
            return (message, line, column);
        }
    }

    (error.to_string(), None, None)
}

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

    #[test]
    fn test_path_matches() {
        assert!(path_matches("hello", "hello"));
        assert!(path_matches("users/:id", "users/123"));
        assert!(path_matches("api/v1/:resource", "api/v1/posts"));
        assert!(!path_matches("hello", "world"));
        assert!(!path_matches("users/:id", "users/123/posts"));
    }

    #[test]
    fn test_extract_params() {
        let params = extract_path_params("users/:id", "users/123");
        assert_eq!(params.get("id").unwrap(), "123");

        let params = extract_path_params("posts/:id/comments/:cid", "posts/10/comments/5");
        assert_eq!(params.get("id").unwrap(), "10");
        assert_eq!(params.get("cid").unwrap(), "5");
    }

    #[test]
    fn test_sanitize_path() {
        assert_eq!(sanitize_path_to_key("hello").as_deref(), Some("hello"));
        assert_eq!(
            sanitize_path_to_key("users/:id").as_deref(),
            Some("users__id")
        );
        assert_eq!(
            sanitize_path_to_key("/api/test").as_deref(),
            Some("api_test")
        );
        assert_eq!(sanitize_path_to_key("foo/../bar"), None);
        assert_eq!(sanitize_path_to_key("../etc/passwd"), None);
    }
}