use std::collections::HashMap;
use std::hash::{DefaultHasher, Hash, Hasher};
use std::path::PathBuf;
use std::sync::Arc;

use axum::extract::{Path, Query, State};
use axum::http::header::HeaderValue;
use axum::http::{HeaderMap, StatusCode};
use axum::response::{IntoResponse, Response};
use axum::Json;
use serde_json::{json, Value};
use tokio::sync::RwLock;

use crate::referrers::{Referrer, ReferrerIndex};
use crate::schema::types::{SchemaDefinition, VALID_FIELD_TYPES};
use crate::schema::validator;
use crate::schema::SchemaRegistry;
use crate::store::query::{QueryParams, StatusFilter, entry_is_draft};
use crate::store::ContentStore;
use crate::store::slug;

use super::auth;
use super::cache::{self, ContentCache, TransformCache};
use super::error::ApiError;
use super::response::{collapse_asset_urls, expand_asset_urls, format_references, parse_resolve};
use super::webhooks;

/// Shared application state. Registry and OpenAPI spec are behind RwLock for hot-reload.
/// Store is selected via RUSTYCMS_STORE=file|sqlite.
/// If api_keys is set (RUSTYCMS_API_KEY or RUSTYCMS_API_KEYS), write operations require a matching key with the right role.
/// When locales is set (RUSTYCMS_LOCALES e.g. "de,en"), API accepts _locale query param.
pub struct AppState {
    pub registry: Arc<RwLock<SchemaRegistry>>,
    pub store: Arc<dyn ContentStore>,
    pub openapi_spec: Arc<RwLock<serde_json::Value>>,
    /// Path to types directory (e.g. ./types) for schema file writes.
    pub types_dir: PathBuf,
    /// API keys with roles (from RUSTYCMS_API_KEYS or RUSTYCMS_API_KEY).
    pub api_keys: Option<super::auth::ApiKeys>,
    pub cache: Arc<ContentCache>,
    pub transform_cache: Arc<TransformCache>,
    pub http_client: reqwest::Client,
    /// If set, first element is default locale. Enables content/{locale}/{collection}/.
    pub locales: Option<Vec<String>>,
    /// Path to the assets directory (e.g. ./content/assets) for image storage.
    pub assets_dir: PathBuf,
    /// Public base URL (e.g. https://api.example.com). Used to expand relative /api/assets/ paths.
    pub base_url: String,
    /// Webhook URLs to POST on content/asset/schema changes (from RUSTYCMS_WEBHOOKS).
    pub webhook_urls: Vec<String>,
    /// When set (RUSTYCMS_ENVIRONMENTS), content/assets are per-environment (e.g. production, staging).
    pub environments: Option<Vec<String>>,
    /// Store per environment when environments is set. Key = env name.
    pub stores: Option<HashMap<String, Arc<dyn ContentStore>>>,
    /// Assets dir per environment when environments is set. Key = env name.
    pub assets_dirs: Option<HashMap<String, PathBuf>>,
    /// Reverse index for "who references this entry?". Updated on create/update/delete. None when environments are used.
    pub referrer_index: Option<Arc<RwLock<ReferrerIndex>>>,
    /// Path to persist referrer index (e.g. content/_referrers.json).
    pub referrer_index_path: Option<PathBuf>,
}

impl AppState {
    /// Resolve environment from query _environment. Default = first in list.
    pub fn effective_environment(&self, params: &HashMap<String, String>) -> String {
        let requested = params.get("_environment").map(|s| s.trim()).filter(|s| !s.is_empty());
        self.effective_environment_from_param(requested)
    }

    /// Resolve environment from optional _environment value (e.g. from asset query params).
    pub fn effective_environment_from_param(&self, env_param: Option<&str>) -> String {
        let Some(ref list) = self.environments else {
            return "default".to_string();
        };
        if list.is_empty() {
            return "default".to_string();
        }
        match env_param {
            Some(env) if list.iter().any(|e| e == env) => env.to_string(),
            _ => list[0].clone(),
        }
    }

    /// Store for the given environment. When environments is off, returns the single store.
    pub fn store_for(&self, env: &str) -> Arc<dyn ContentStore> {
        self.stores
            .as_ref()
            .and_then(|m| m.get(env).cloned())
            .unwrap_or_else(|| Arc::clone(&self.store))
    }

    /// Assets directory for the given environment.
    pub fn assets_dir_for(&self, env: &str) -> PathBuf {
        self.assets_dirs
            .as_ref()
            .and_then(|m| m.get(env).cloned())
            .unwrap_or_else(|| self.assets_dir.clone())
    }
}

/// Resolve effective locale from query _locale and state.locales. Returns None when i18n is off.
pub fn effective_locale(params: &HashMap<String, String>, locales: Option<&[String]>) -> Option<String> {
    let locales = locales?;
    if locales.is_empty() {
        return None;
    }
    let requested = params.get("_locale").map(|s| s.trim()).filter(|s| !s.is_empty());
    match requested {
        Some(loc) if locales.iter().any(|l| l == loc) => Some(loc.to_string()),
        _ => Some(locales[0].clone()),
    }
}

// ---------------------------------------------------------------------------
// GET /health
// ---------------------------------------------------------------------------

pub async fn health() -> (StatusCode, Json<Value>) {
    (StatusCode::OK, Json(json!({ "status": "ok" })))
}

// ---------------------------------------------------------------------------
// GET /api/collections
// ---------------------------------------------------------------------------

pub async fn list_collections(
    State(state): State<Arc<AppState>>,
) -> Json<Value> {
    let collections: Vec<Value> = {
        let registry = state.registry.read().await;
        registry
            .list()
            .iter()
            .filter(|(_, schema)| !schema.reusable)
            .map(|(name, schema)| {
                json!({
                    "name": name,
                    "field_count": schema.fields.len(),
                    "extends": schema.extends,
                    "strict": schema.strict,
                    "description": schema.description,
                    "tags": schema.tags,
                    "category": schema.category,
                })
            })
            .collect()
    };

    Json(json!({ "collections": collections }))
}

// ---------------------------------------------------------------------------
// POST /api/schemas – Create new content type (schema)
// ---------------------------------------------------------------------------

/// Content type name: lowercase letters, digits, underscore only.
fn is_valid_schema_name(name: &str) -> bool {
    !name.is_empty()
        && name.len() <= 64
        && name
            .chars()
            .all(|c| c.is_ascii_lowercase() || c.is_ascii_digit() || c == '_')
}

pub async fn create_schema(
    State(state): State<Arc<AppState>>,
    headers: HeaderMap,
    Json(schema): Json<SchemaDefinition>,
) -> Result<(StatusCode, Json<Value>), ApiError> {
    if let Some(ref keys) = state.api_keys {
        keys.require(&headers, auth::Permission::SchemasWrite)?;
    }

    if !is_valid_schema_name(&schema.name) {
        return Err(ApiError::BadRequest(
            "name: lowercase letters, digits and underscore only, max 64 chars".to_string(),
        ));
    }

    if schema.reusable {
        return Err(ApiError::BadRequest(
            "reusable must not be true for new content types".to_string(),
        ));
    }

    let mut errors = Vec::new();
    for (field_name, fd) in &schema.fields {
        if !VALID_FIELD_TYPES.contains(&fd.field_type.as_str()) {
            errors.push(format!(
                "Field '{}': unknown type '{}'",
                field_name, fd.field_type
            ));
        }
    }
    if !errors.is_empty() {
        return Err(ApiError::ValidationFailed(errors));
    }

    let path = state.types_dir.join(format!("{}.json", schema.name));
    let contents = serde_json::to_string_pretty(&schema).map_err(|e| ApiError::Internal(e.to_string()))?;

    tokio::fs::write(&path, contents)
        .await
        .map_err(|e| ApiError::Internal(format!("Failed to write schema file: {}", e)))?;

    state
        .store
        .ensure_collection_dir(&schema.name)
        .await
        .map_err(ApiError::from)?;

    tracing::info!("Schema created: {} ({})", schema.name, path.display());

    webhooks::fire(
        state.http_client.clone(),
        &state.webhook_urls,
        json!({
            "event": webhooks::EVENT_SCHEMA_CREATED,
            "name": schema.name,
        }),
    );

    Ok((
        StatusCode::CREATED,
        Json(serde_json::to_value(&schema).unwrap()),
    ))
}

// ---------------------------------------------------------------------------
// PUT /api/schemas/:name – update existing schema
// ---------------------------------------------------------------------------

pub async fn update_schema(
    State(state): State<Arc<AppState>>,
    Path(name): Path<String>,
    headers: HeaderMap,
    Json(schema): Json<SchemaDefinition>,
) -> Result<Json<Value>, ApiError> {
    if let Some(ref keys) = state.api_keys {
        keys.require(&headers, auth::Permission::SchemasWrite)?;
    }

    if name != schema.name {
        return Err(ApiError::BadRequest(
            "Path name must match schema.name".to_string(),
        ));
    }
    if !is_valid_schema_name(&schema.name) {
        return Err(ApiError::BadRequest(
            "name: lowercase letters, digits and underscore only, max 64 chars".to_string(),
        ));
    }

    let mut errors = Vec::new();
    for (field_name, fd) in &schema.fields {
        if !VALID_FIELD_TYPES.contains(&fd.field_type.as_str()) {
            errors.push(format!(
                "Field '{}': unknown type '{}'",
                field_name, fd.field_type
            ));
        }
    }
    if !errors.is_empty() {
        return Err(ApiError::ValidationFailed(errors));
    }

    let json5_path = state.types_dir.join(format!("{}.json5", name));
    let json_path = state.types_dir.join(format!("{}.json", name));
    let path = if json5_path.exists() {
        json5_path
    } else if json_path.exists() {
        json_path
    } else {
        return Err(ApiError::NotFound(format!("Schema '{}' not found", name)));
    };

    let contents = serde_json::to_string_pretty(&schema).map_err(|e| ApiError::Internal(e.to_string()))?;
    tokio::fs::write(&path, contents)
        .await
        .map_err(|e| ApiError::Internal(format!("Failed to write schema file: {}", e)))?;

    tracing::info!("Schema updated: {} ({})", name, path.display());

    webhooks::fire(
        state.http_client.clone(),
        &state.webhook_urls,
        json!({
            "event": webhooks::EVENT_SCHEMA_UPDATED,
            "name": name,
        }),
    );

    Ok(Json(serde_json::to_value(&schema).unwrap()))
}

// ---------------------------------------------------------------------------
// DELETE /api/schemas/:name – delete schema file
// ---------------------------------------------------------------------------

pub async fn delete_schema(
    State(state): State<Arc<AppState>>,
    Path(name): Path<String>,
    headers: HeaderMap,
) -> Result<StatusCode, ApiError> {
    if let Some(ref keys) = state.api_keys {
        keys.require(&headers, auth::Permission::SchemasWrite)?;
    }

    if !is_valid_schema_name(&name) {
        return Err(ApiError::BadRequest("Invalid schema name".to_string()));
    }

    let json5_path = state.types_dir.join(format!("{}.json5", name));
    let json_path = state.types_dir.join(format!("{}.json", name));
    let mut deleted = false;
    if json5_path.exists() {
        tokio::fs::remove_file(&json5_path)
            .await
            .map_err(|e| ApiError::Internal(format!("Failed to delete schema file: {}", e)))?;
        deleted = true;
        tracing::info!("Schema deleted: {} ({})", name, json5_path.display());
    }
    if json_path.exists() {
        tokio::fs::remove_file(&json_path)
            .await
            .map_err(|e| ApiError::Internal(format!("Failed to delete schema file: {}", e)))?;
        deleted = true;
        tracing::info!("Schema deleted: {} ({})", name, json_path.display());
    }
    if !deleted {
        return Err(ApiError::NotFound(format!("Schema '{}' not found", name)));
    }

    webhooks::fire(
        state.http_client.clone(),
        &state.webhook_urls,
        json!({
            "event": webhooks::EVENT_SCHEMA_DELETED,
            "name": name,
        }),
    );

    Ok(StatusCode::NO_CONTENT)
}

// ---------------------------------------------------------------------------
// GET /api/collections/:collection
// ---------------------------------------------------------------------------

pub async fn get_collection_schema(
    State(state): State<Arc<AppState>>,
    Path(collection): Path<String>,
) -> Result<Json<Value>, ApiError> {
    let schema = {
        let registry = state.registry.read().await;
        registry
            .get(&collection)
            .ok_or_else(|| ApiError::NotFound(format!("Collection '{}' not found", collection)))?
            .clone()
    };
    if schema.reusable {
        return Err(ApiError::NotFound(format!(
            "Collection '{}' is a reusable partial, not a content collection",
            collection
        )));
    }

    Ok(Json(serde_json::to_value(schema).unwrap()))
}

// ---------------------------------------------------------------------------
// GET /api/schemas – list all type names (including reusable, for extends dropdown)
// ---------------------------------------------------------------------------

pub async fn list_schema_names(
    State(state): State<Arc<AppState>>,
) -> Json<Value> {
    let names: Vec<String> = {
        let registry = state.registry.read().await;
        registry.names()
    };
    Json(json!({ "names": names }))
}

// ---------------------------------------------------------------------------
// GET /api/schemas/:name/raw – schema as stored on disk (own fields + extends, unresolved)
// ---------------------------------------------------------------------------

pub async fn get_schema_raw(
    State(state): State<Arc<AppState>>,
    Path(name): Path<String>,
) -> Result<Json<Value>, ApiError> {
    let path_json5 = state.types_dir.join(format!("{}.json5", name));
    let path_json = state.types_dir.join(format!("{}.json", name));
    let path = if path_json5.exists() {
        path_json5
    } else if path_json.exists() {
        path_json
    } else {
        return Err(ApiError::NotFound(format!("Schema '{}' not found", name)));
    };
    let content = tokio::fs::read_to_string(&path)
        .await
        .map_err(|e| ApiError::Internal(format!("Failed to read schema file: {}", e)))?;
    let schema: SchemaDefinition = json5::from_str(&content)
        .map_err(|e| ApiError::BadRequest(format!("Invalid schema file: {}", e)))?;
    if schema.name != name {
        return Err(ApiError::BadRequest(format!(
            "Schema file name mismatch: expected '{}', got '{}'",
            name, schema.name
        )));
    }
    Ok(Json(serde_json::to_value(schema).unwrap()))
}

// ---------------------------------------------------------------------------
// GET /api/collections/:collection/slug-check?slug=xxx&exclude=yyy&_locale=zzz
// ---------------------------------------------------------------------------

/// Response: { valid, normalized, available, error? }
/// - valid: slug format is ok (after normalize)
/// - normalized: normalized slug string
/// - available: not used by another entry (or equals exclude when editing)
pub async fn slug_check(
    State(state): State<Arc<AppState>>,
    Path(collection): Path<String>,
    Query(params): Query<HashMap<String, String>>,
) -> Result<Json<Value>, ApiError> {
    let slug_raw = params
        .get("slug")
        .filter(|s| !s.trim().is_empty())
        .cloned()
        .unwrap_or_default();

    let valid = slug::validate_slug(&slug_raw).is_ok();
    let normalized = slug::normalize_slug(&slug_raw);

    let error = if !valid {
        if slug_raw.is_empty() {
            Some("Slug must not be empty.".to_string())
        } else if normalized.is_empty() {
            Some("Slug contains invalid characters. Allowed: lowercase, digits, hyphens (a-z, 0-9, -).".to_string())
        } else if slug_raw.contains('/') || slug_raw.contains('\\') || slug_raw.contains("..") {
            Some("Slug must not contain path characters (/, \\, ..).".to_string())
        } else {
            Some("Invalid slug.".to_string())
        }
    } else {
        None
    };

    let available = if !valid || normalized.is_empty() {
        false
    } else {
        let exclude = params.get("exclude").map(|s| s.trim()).filter(|s| !s.is_empty());
        let locale = effective_locale(&params, state.locales.as_deref());
        let locale_ref = locale.as_deref();

        let registry = state.registry.read().await;
        let _schema = match registry.get(&collection) {
            Some(s) if !s.reusable => s,
            _ => {
                return Ok(Json(json!({
                    "valid": valid,
                    "normalized": normalized,
                    "available": false,
                    "error": "Collection not found."
                })));
            }
        };
        drop(registry);

        if exclude.as_deref() == Some(normalized.as_str()) {
            true
        } else {
            let env = state.effective_environment(&params);
            let store = state.store_for(&env);
            let exists = store
                .get(&collection, &normalized, locale_ref)
                .await
                .map_err(ApiError::from)?
                .is_some();
            !exists
        }
    };

    Ok(Json(json!({
        "valid": valid,
        "normalized": normalized,
        "available": available,
        "error": error,
    })))
}

// ---------------------------------------------------------------------------
// GET /api/content/:collection
// ---------------------------------------------------------------------------

pub async fn list_entries(
    State(state): State<Arc<AppState>>,
    Path(collection): Path<String>,
    Query(params): Query<HashMap<String, String>>,
    headers: HeaderMap,
) -> Result<Json<Value>, ApiError> {
    let registry = state.registry.read().await;
    let schema = registry
        .get(&collection)
        .ok_or_else(|| ApiError::NotFound(format!("Collection '{}' not found", collection)))?;
    if schema.reusable {
        return Err(ApiError::NotFound(format!(
            "Collection '{}' is a reusable partial, not a content collection",
            collection
        )));
    }

    let env = state.effective_environment(&params);
    let store = state.store_for(&env);
    let locale = effective_locale(&params, state.locales.as_deref());
    let locale_ref = locale.as_deref();

    // Apply schema default sort when client does not send _sort
    let mut list_params = params.clone();
    if !list_params.contains_key("_sort") {
        if let Some(ref sort) = schema.default_sort {
            list_params.insert("_sort".to_string(), sort.clone());
            list_params.insert(
                "_order".to_string(),
                schema.default_order.clone().unwrap_or_else(|| "desc".to_string()),
            );
        }
    }

    let mut keys: Vec<_> = list_params.keys().collect();
    keys.sort();
    let mut hasher = DefaultHasher::new();
    for k in &keys {
        k.hash(&mut hasher);
        list_params.get(k.as_str()).unwrap().hash(&mut hasher);
    }
    let query_hash = hasher.finish();
    let cache_key = cache::list_cache_key(&env, &collection, query_hash, locale_ref);
    if let Some(cached) = state.cache.get(&cache_key).await {
        return Ok(Json(cached));
    }

    let entries = store.list(&collection, locale_ref).await.map_err(ApiError::from)?;

    let resolve = parse_resolve(list_params.get("_resolve").map(|s| s.as_str()));
    // When API keys are enabled but request has no valid key, show only published entries.
    let status_override = if state
        .api_keys
        .as_ref()
        .map(|k| k.is_enabled() && !k.is_authenticated(&headers))
        .unwrap_or(false)
    {
        Some(StatusFilter::Published)
    } else {
        None
    };
    let query = QueryParams::from_map_with_status(list_params, status_override);
    let mut result = query.apply(entries);

    for item in result.items.iter_mut() {
        *item = format_references(
            std::mem::take(item),
            schema,
            store.as_ref(),
            resolve.as_ref(),
            locale_ref,
            Some(&*registry),
        )
        .await;
        if resolve.is_some() {
            expand_asset_urls(item, &state.base_url);
        }
    }

    let response_value = serde_json::to_value(&result).unwrap();
    state.cache.set(cache_key, response_value.clone()).await;
    Ok(Json(response_value))
}

// ---------------------------------------------------------------------------
// GET /api/content/:collection/:slug
// ---------------------------------------------------------------------------

pub async fn get_entry(
    State(state): State<Arc<AppState>>,
    Path((collection, slug)): Path<(String, String)>,
    Query(params): Query<HashMap<String, String>>,
    headers: HeaderMap,
) -> Result<Response, ApiError> {
    let registry = state.registry.read().await;
    let schema = registry
        .get(&collection)
        .ok_or_else(|| ApiError::NotFound(format!("Collection '{}' not found", collection)))?;
    if schema.reusable {
        return Err(ApiError::NotFound(format!(
            "Collection '{}' is a reusable partial, not a content collection",
            collection
        )));
    }

    let env = state.effective_environment(&params);
    let store = state.store_for(&env);
    let locale = effective_locale(&params, state.locales.as_deref());
    let locale_ref = locale.as_deref();

    let resolve_key = params.get("_resolve").map(|s| s.as_str()).unwrap_or("");
    let cache_key = cache::entry_cache_key(&env, &collection, &slug, resolve_key, locale_ref);
    if let Some(ref cached) = state.cache.get(&cache_key).await {
        // Don't serve cached draft to unauthenticated requests.
        let is_authenticated = state
            .api_keys
            .as_ref()
            .map(|k| k.is_authenticated(&headers))
            .unwrap_or(true);
        if !is_authenticated && entry_is_draft(cached) {
            // Fall through to load from store (will 404 if draft).
        } else {
            let json_str = serde_json::to_string(cached).unwrap_or_default();
            let mut hasher = DefaultHasher::new();
            json_str.hash(&mut hasher);
            let etag_plain = format!("\"{:016x}\"", hasher.finish());
            let etag_header =
                HeaderValue::from_str(&etag_plain).unwrap_or_else(|_| HeaderValue::from_static("\"0\""));
            let if_none_match = headers
                .get("if-none-match")
                .and_then(|v| v.to_str().ok())
                .map(|s| s.trim().trim_matches('"').to_string());
            if if_none_match.as_deref() == Some(etag_plain.trim_matches('"')) {
                return Ok((
                    StatusCode::NOT_MODIFIED,
                    [("ETag", etag_header.clone())],
                )
                    .into_response());
            }
            return Ok((
                StatusCode::OK,
                [("ETag", etag_header)],
                Json(cached.clone()),
            )
                .into_response());
        }
    }

    let entry = store
        .get(&collection, &slug, locale_ref)
        .await
        .map_err(ApiError::from)?
        .ok_or_else(|| {
            ApiError::NotFound(format!("Entry '{}' not found in '{}'", slug, collection))
        })?;

    // When API keys are enabled and request has no valid key, hide draft entries (return 404).
    if state
        .api_keys
        .as_ref()
        .map(|k| k.is_enabled() && !k.is_authenticated(&headers))
        .unwrap_or(false)
        && entry_is_draft(&entry)
    {
        return Err(ApiError::NotFound(format!(
            "Entry '{}' not found in '{}'",
            slug, collection
        )));
    }

    let resolve = parse_resolve(params.get("_resolve").map(|s| s.as_str()));
    let mut formatted = format_references(
        entry,
        schema,
        store.as_ref(),
        resolve.as_ref(),
        locale_ref,
        Some(&*registry),
    )
    .await;
    if resolve.is_some() {
        expand_asset_urls(&mut formatted, &state.base_url);
    }

    // Only cache published entries so unauthenticated requests never see cached drafts.
    if !entry_is_draft(&formatted) {
        state.cache.set(cache_key.clone(), formatted.clone()).await;
    }

    let json_str = serde_json::to_string(&formatted).unwrap_or_default();
    let mut hasher = DefaultHasher::new();
    json_str.hash(&mut hasher);
    let etag_plain = format!("\"{:016x}\"", hasher.finish());
    let etag_header =
        HeaderValue::from_str(&etag_plain).unwrap_or_else(|_| HeaderValue::from_static("\"0\""));

    let if_none_match = headers
        .get("if-none-match")
        .and_then(|v| v.to_str().ok())
        .map(|s| s.trim().trim_matches('"').to_string());
    if if_none_match.as_deref() == Some(etag_plain.trim_matches('"')) {
        return Ok((
            StatusCode::NOT_MODIFIED,
            [("ETag", etag_header.clone())],
        )
            .into_response());
    }

    Ok((
        StatusCode::OK,
        [("ETag", etag_header)],
        Json(formatted),
    )
        .into_response())
}

// ---------------------------------------------------------------------------
// POST /api/content/:collection
// ---------------------------------------------------------------------------

pub async fn create_entry(
    State(state): State<Arc<AppState>>,
    Path(collection): Path<String>,
    Query(params): Query<HashMap<String, String>>,
    headers: HeaderMap,
    Json(mut body): Json<Value>,
) -> Result<(StatusCode, Json<Value>), ApiError> {
    if let Some(ref keys) = state.api_keys {
        keys.require(&headers, auth::Permission::ContentWrite)?;
    }

    let env = state.effective_environment(&params);
    let store = state.store_for(&env);
    let locale = effective_locale(&params, state.locales.as_deref());
    let locale_ref = locale.as_deref();

    let schema = {
        let registry = state.registry.read().await;
        registry
            .get(&collection)
            .ok_or_else(|| ApiError::NotFound(format!("Collection '{}' not found", collection)))?
            .clone()
    };
    if schema.reusable {
        return Err(ApiError::NotFound(format!(
            "Collection '{}' is a reusable partial, not a content collection",
            collection
        )));
    }

    // Extract and normalize _slug from body
    let slug_raw = body
        .as_object()
        .and_then(|o| o.get("_slug"))
        .and_then(|v| v.as_str())
        .map(|s| s.to_string())
        .ok_or_else(|| ApiError::BadRequest("Field '_slug' is required".to_string()))?;
    slug::validate_slug(&slug_raw).map_err(ApiError::BadRequest)?;
    let slug = slug::normalize_slug(&slug_raw);

    // Remove _slug from content data; validate and default _status (publish/draft)
    if let Some(obj) = body.as_object_mut() {
        obj.remove("_slug");
        if let Some(v) = obj.get("_status") {
            let s = v.as_str().unwrap_or("");
            if s != "draft" && s != "published" {
                return Err(ApiError::BadRequest(
                    "Field '_status' must be \"draft\" or \"published\"".to_string(),
                ));
            }
        } else {
            obj.insert("_status".to_string(), Value::String("published".to_string()));
        }
    }

    // Apply defaults and auto-generated values
    validator::apply_defaults(&schema, &mut body);

    // Normalize reference arrays: objects with _slug → slug strings (so clients can send resolved refs)
    validator::normalize_reference_arrays(&schema, &mut body);
    validator::normalize_multi_select(&schema, &mut body);

    // Validate against schema (type checks, constraints, strict mode, …)
    let errors = validator::validate_content(&schema, &body);
    if !errors.is_empty() {
        let messages: Vec<String> = errors.iter().map(|e| e.to_string()).collect();
        return Err(ApiError::ValidationFailed(messages));
    }

    // Unique constraint check (within same locale)
    let entries = store.list(&collection, locale_ref).await.map_err(ApiError::from)?;
    let unique_errors = validator::validate_unique(&schema, &body, None, &entries);
    if !unique_errors.is_empty() {
        let messages: Vec<String> = unique_errors.iter().map(|e| e.to_string()).collect();
        return Err(ApiError::ValidationFailed(messages));
    }

    // Reference validation (blocking: we need sync closure; use tokio::task::block_in_place or spawn)
    let ref_errors = validator::validate_references(&schema, &body, &|coll, s| {
        let store_ref = Arc::clone(&store);
        tokio::task::block_in_place(|| {
            tokio::runtime::Handle::current().block_on(async move {
                store_ref.get(coll, s, locale_ref).await.ok().flatten().is_some()
            })
        })
    });
    if !ref_errors.is_empty() {
        let messages: Vec<String> = ref_errors.iter().map(|e| e.to_string()).collect();
        return Err(ApiError::ValidationFailed(messages));
    }

    // Normalize absolute asset URLs → relative before persisting
    collapse_asset_urls(&mut body, &state.base_url);

    // Persist to filesystem
    store
        .create(&collection, &slug, &body, locale_ref)
        .await
        .map_err(ApiError::from)?;
    state.cache.invalidate_collection(&env, &collection).await;

    // Update referrer index: this entry references (ref_coll, ref_slug)
    if let (Some(ref idx), Some(ref path)) = (&state.referrer_index, &state.referrer_index_path) {
        let refs = validator::extract_references(&schema, &body);
        let referrer = Referrer {
            collection: collection.clone(),
            slug: slug.clone(),
            field: String::new(), // we add per (ref_coll, ref_slug, field) below
            locale: locale_ref.map(str::to_string),
        };
        let mut index = idx.write().await;
        for (ref_coll, ref_slug, field) in refs {
            let mut r = referrer.clone();
            r.field = field;
            index.add_referrer(&ref_coll, &ref_slug, r);
        }
        drop(index);
        if let Err(e) = idx.read().await.save(path) {
            tracing::warn!("Failed to save referrer index: {}", e);
        }
    }

    // Return created entry (with reference format)
    let entry = store
        .get(&collection, &slug, locale_ref)
        .await
        .map_err(ApiError::from)?
        .unwrap();
    let formatted = format_references(
        entry,
        &schema,
        store.as_ref(),
        None,
        locale_ref,
        None,
    )
    .await;

    webhooks::fire(
        state.http_client.clone(),
        &state.webhook_urls,
        json!({
            "event": webhooks::EVENT_CONTENT_CREATED,
            "collection": collection,
            "slug": slug,
            "locale": locale_ref,
            "environment": env,
        }),
    );

    Ok((StatusCode::CREATED, Json(formatted)))
}

// ---------------------------------------------------------------------------
// PUT /api/content/:collection/:slug
// ---------------------------------------------------------------------------

pub async fn update_entry(
    State(state): State<Arc<AppState>>,
    Path((collection, slug)): Path<(String, String)>,
    Query(params): Query<HashMap<String, String>>,
    headers: HeaderMap,
    Json(mut body): Json<Value>,
) -> Result<Json<Value>, ApiError> {
    if let Some(ref keys) = state.api_keys {
        keys.require(&headers, auth::Permission::ContentWrite)?;
    }

    let env = state.effective_environment(&params);
    let store = state.store_for(&env);
    let locale = effective_locale(&params, state.locales.as_deref());
    let locale_ref = locale.as_deref();

    let schema = {
        let registry = state.registry.read().await;
        registry
            .get(&collection)
            .ok_or_else(|| ApiError::NotFound(format!("Collection '{}' not found", collection)))?
            .clone()
    };
    if schema.reusable {
        return Err(ApiError::NotFound(format!(
            "Collection '{}' is a reusable partial, not a content collection",
            collection
        )));
    }

    // Remove _slug if present; validate _status when present
    if let Some(obj) = body.as_object_mut() {
        obj.remove("_slug");
        if let Some(v) = obj.get("_status") {
            let s = v.as_str().unwrap_or("");
            if s != "draft" && s != "published" {
                return Err(ApiError::BadRequest(
                    "Field '_status' must be \"draft\" or \"published\"".to_string(),
                ));
            }
        }
    }

    // Normalize reference arrays: objects with _slug → slug strings (so clients can send resolved refs)
    validator::normalize_reference_arrays(&schema, &mut body);
    validator::normalize_multi_select(&schema, &mut body);

    // Load existing content for readonly check
    let existing = store
        .get(&collection, &slug, locale_ref)
        .await
        .map_err(ApiError::from)?
        .ok_or_else(|| {
            ApiError::NotFound(format!("Entry '{}' not found in '{}'", slug, collection))
        })?;

    // Readonly violation check
    let readonly_errors = validator::check_readonly_violations(&schema, &existing, &body);
    if !readonly_errors.is_empty() {
        let messages: Vec<String> = readonly_errors.iter().map(|e| e.to_string()).collect();
        return Err(ApiError::ValidationFailed(messages));
    }

    // Validate against schema
    let errors = validator::validate_content(&schema, &body);
    if !errors.is_empty() {
        let messages: Vec<String> = errors.iter().map(|e| e.to_string()).collect();
        return Err(ApiError::ValidationFailed(messages));
    }

    // Unique constraint check (exclude self, within same locale)
    let entries = store.list(&collection, locale_ref).await.map_err(ApiError::from)?;
    let unique_errors = validator::validate_unique(&schema, &body, Some(&slug), &entries);
    if !unique_errors.is_empty() {
        let messages: Vec<String> = unique_errors.iter().map(|e| e.to_string()).collect();
        return Err(ApiError::ValidationFailed(messages));
    }

    // Reference validation
    let ref_errors = validator::validate_references(&schema, &body, &|coll, s| {
        let store_ref = Arc::clone(&store);
        tokio::task::block_in_place(|| {
            tokio::runtime::Handle::current().block_on(async move {
                store_ref.get(coll, s, locale_ref).await.ok().flatten().is_some()
            })
        })
    });
    if !ref_errors.is_empty() {
        let messages: Vec<String> = ref_errors.iter().map(|e| e.to_string()).collect();
        return Err(ApiError::ValidationFailed(messages));
    }

    // Normalize absolute asset URLs → relative before persisting
    collapse_asset_urls(&mut body, &state.base_url);

    // Persist to filesystem
    store
        .update(&collection, &slug, &body, locale_ref)
        .await
        .map_err(ApiError::from)?;
    state.cache.invalidate_collection(&env, &collection).await;

    // Update referrer index: remove old refs from this entry, add new refs
    if let (Some(ref idx), Some(ref path)) = (&state.referrer_index, &state.referrer_index_path) {
        let mut index = idx.write().await;
        index.remove_all_referrers_from(&collection, &slug);
        let refs = validator::extract_references(&schema, &body);
        let referrer = Referrer {
            collection: collection.clone(),
            slug: slug.clone(),
            field: String::new(),
            locale: locale_ref.map(str::to_string),
        };
        for (ref_coll, ref_slug, field) in refs {
            let mut r = referrer.clone();
            r.field = field;
            index.add_referrer(&ref_coll, &ref_slug, r);
        }
        drop(index);
        if let Err(e) = idx.read().await.save(path) {
            tracing::warn!("Failed to save referrer index: {}", e);
        }
    }

    // Return updated entry (with reference format)
    let entry = store
        .get(&collection, &slug, locale_ref)
        .await
        .map_err(ApiError::from)?
        .unwrap();
    let formatted = format_references(
        entry,
        &schema,
        store.as_ref(),
        None,
        locale_ref,
        None,
    )
    .await;

    webhooks::fire(
        state.http_client.clone(),
        &state.webhook_urls,
        json!({
            "event": webhooks::EVENT_CONTENT_UPDATED,
            "collection": collection,
            "slug": slug,
            "locale": locale_ref,
            "environment": env,
        }),
    );

    Ok(Json(formatted))
}

// ---------------------------------------------------------------------------
// GET /api/content/:collection/:slug/referrers
// ---------------------------------------------------------------------------
pub async fn get_referrers(
    State(state): State<Arc<AppState>>,
    Path((collection, slug)): Path<(String, String)>,
) -> Result<Json<Value>, ApiError> {
    let referrers = match &state.referrer_index {
        Some(idx) => idx.read().await.get_referrers(&collection, &slug),
        None => vec![],
    };
    Ok(Json(
        serde_json::to_value(&referrers).unwrap_or_else(|_| json!([])),
    ))
}

// ---------------------------------------------------------------------------
// GET /api/locales
// ---------------------------------------------------------------------------
pub async fn list_locales(State(state): State<Arc<AppState>>) -> Json<Value> {
    match &state.locales {
        Some(locales) if !locales.is_empty() => Json(json!({
            "locales": locales,
            "default": locales[0],
        })),
        _ => Json(json!({
            "locales": [],
            "default": null,
        })),
    }
}

// ---------------------------------------------------------------------------
// DELETE /api/content/:collection/:slug
// ---------------------------------------------------------------------------

pub async fn delete_entry(
    State(state): State<Arc<AppState>>,
    Path((collection, slug)): Path<(String, String)>,
    Query(params): Query<HashMap<String, String>>,
    headers: HeaderMap,
) -> Result<StatusCode, ApiError> {
    if let Some(ref keys) = state.api_keys {
        keys.require(&headers, auth::Permission::ContentWrite)?;
    }

    let env = state.effective_environment(&params);
    let store = state.store_for(&env);
    let locale = effective_locale(&params, state.locales.as_deref());
    let locale_ref = locale.as_deref();

    let schema = {
        let registry = state.registry.read().await;
        registry
            .get(&collection)
            .ok_or_else(|| ApiError::NotFound(format!("Collection '{}' not found", collection)))?
            .clone()
    };
    if schema.reusable {
        return Err(ApiError::NotFound(format!(
            "Collection '{}' is a reusable partial, not a content collection",
            collection
        )));
    }

    // Update referrer index: remove this entry from every (ref_coll, ref_slug) it referenced; and remove key (collection, slug)
    if let (Some(ref idx), Some(ref path)) = (&state.referrer_index, &state.referrer_index_path) {
        let mut index = idx.write().await;
        index.remove_all_referrers_from(&collection, &slug);
        index.remove_referenced_entry(&collection, &slug);
        drop(index);
        if let Err(e) = idx.read().await.save(path) {
            tracing::warn!("Failed to save referrer index: {}", e);
        }
    }

    store
        .delete(&collection, &slug, locale_ref)
        .await
        .map_err(ApiError::from)?;
    state.cache.invalidate_collection(&env, &collection).await;

    webhooks::fire(
        state.http_client.clone(),
        &state.webhook_urls,
        json!({
            "event": webhooks::EVENT_CONTENT_DELETED,
            "collection": collection,
            "slug": slug,
            "locale": locale_ref,
            "environment": env,
        }),
    );

    Ok(StatusCode::NO_CONTENT)
}