import asyncio
import json
import logging
from uuid import UUID

from fastapi import APIRouter, HTTPException
from lfx.base.models.unified_models import get_model_provider_variable_mapping, validate_model_provider_key
from sqlalchemy.exc import NoResultFound

from langflow.api.utils import CurrentActiveUser, DbSession
from langflow.api.v1.models import (
    DISABLED_MODELS_VAR,
    ENABLED_MODELS_VAR,
    get_model_names_for_provider,
    get_provider_from_variable_name,
)
from langflow.api.v1.schemas.deployments import DetectVarsRequest, DetectVarsResponse
from langflow.services.database.models.flow_version.crud import get_flow_version_entries_by_ids
from langflow.services.database.models.variable.model import VariableCreate, VariableRead, VariableUpdate
from langflow.services.deps import get_variable_service
from langflow.services.variable.constants import CREDENTIAL_TYPE, GENERIC_TYPE
from langflow.services.variable.service import DatabaseVariableService

router = APIRouter(prefix="/variables", tags=["Variables"])
model_provider_variable_mapping = get_model_provider_variable_mapping()
logger = logging.getLogger(__name__)


async def _cleanup_model_list_variable(
    variable_service: DatabaseVariableService,
    user_id: UUID,
    variable_name: str,
    models_to_remove: set[str],
    session: DbSession,
) -> None:
    """Remove specified models from a model list variable (disabled or enabled models).

    If all models are removed, the variable is deleted entirely.
    If the variable doesn't exist, this is a no-op.
    """
    try:
        model_list_var = await variable_service.get_variable_object(
            user_id=user_id, name=variable_name, session=session
        )
    except ValueError:
        # Variable doesn't exist, nothing to clean up
        return

    if not model_list_var or not model_list_var.value:
        return

    # Parse current models
    try:
        current_models = set(json.loads(model_list_var.value))
    except (json.JSONDecodeError, TypeError):
        current_models = set()

    # Filter out the provider's models
    filtered_models = current_models - models_to_remove

    # Nothing changed, no update needed
    if filtered_models == current_models:
        return

    if filtered_models:
        # Update with filtered list
        if model_list_var.id is not None:
            await variable_service.update_variable_fields(
                user_id=user_id,
                variable_id=model_list_var.id,
                variable=VariableUpdate(
                    id=model_list_var.id,
                    name=variable_name,
                    value=json.dumps(list(filtered_models)),
                    type=GENERIC_TYPE,
                ),
                session=session,
            )
    else:
        # No models left, delete the variable
        await variable_service.delete_variable(user_id=user_id, name=variable_name, session=session)


async def _cleanup_provider_models(
    variable_service: DatabaseVariableService,
    user_id: UUID,
    provider: str,
    session: DbSession,
) -> None:
    """Clean up disabled and enabled model lists for a deleted provider credential."""
    try:
        provider_models = get_model_names_for_provider(provider)
    except ValueError:
        logger.exception("Provider model retrieval failed")
        return

    # Clean up disabled and enabled models
    await _cleanup_model_list_variable(variable_service, user_id, DISABLED_MODELS_VAR, provider_models, session)
    await _cleanup_model_list_variable(variable_service, user_id, ENABLED_MODELS_VAR, provider_models, session)


@router.post("/", response_model=VariableRead, status_code=201, include_in_schema=False)
async def create_variable(
    *,
    session: DbSession,
    variable: VariableCreate,
    current_user: CurrentActiveUser,
):
    """Create a new variable."""
    variable_service = get_variable_service()
    if not variable.name and not variable.value:
        raise HTTPException(status_code=400, detail="Variable name and value cannot be empty")

    if not variable.name:
        raise HTTPException(status_code=400, detail="Variable name cannot be empty")

    if not variable.value:
        raise HTTPException(status_code=400, detail="Variable value cannot be empty")

    if variable.name in await variable_service.list_variables(user_id=current_user.id, session=session):
        raise HTTPException(status_code=400, detail="Variable name already exists")

    # Check if the variable is a reserved model provider variable
    if variable.name in model_provider_variable_mapping.values():
        provider = get_provider_from_variable_name(variable.name)
        if provider is not None:
            # Validate that the key actually works using the Language Model Service
            # Run validation off the event loop to avoid blocking
            try:
                await asyncio.to_thread(validate_model_provider_key, provider, {variable.name: variable.value})
            except ValueError as e:
                raise HTTPException(status_code=400, detail=str(e)) from e

    try:
        return await variable_service.create_variable(
            user_id=current_user.id,
            name=variable.name,
            value=variable.value,
            default_fields=variable.default_fields or [],
            type_=variable.type or CREDENTIAL_TYPE,
            session=session,
        )
    except Exception as e:
        if isinstance(e, HTTPException):
            raise
        raise HTTPException(status_code=500, detail=str(e)) from e


@router.get("/", response_model=list[VariableRead], status_code=200, include_in_schema=False)
async def read_variables(
    *,
    session: DbSession,
    current_user: CurrentActiveUser,
):
    """Read all variables.

    Model provider credentials are validated when they are created or updated,
    not on every read. This avoids latency from external API calls on read operations.

    Returns a list of variables.
    """
    variable_service = get_variable_service()
    if not isinstance(variable_service, DatabaseVariableService):
        msg = "Variable service is not an instance of DatabaseVariableService"
        raise TypeError(msg)
    try:
        all_variables = await variable_service.get_all(user_id=current_user.id, session=session)

        # Filter out internal variables (those starting and ending with __)
        filtered_variables = [
            var for var in all_variables if not (var.name and var.name.startswith("__") and var.name.endswith("__"))
        ]

        # Mark model provider credentials - validation status is based on existence
        # (actual validation happens on create/update)
        for var in filtered_variables:
            if var.name and var.name in model_provider_variable_mapping.values() and var.type == CREDENTIAL_TYPE:
                # Credential exists and was validated on save
                var.is_valid = True
                var.validation_error = None
            else:
                # Not a model provider credential
                var.is_valid = None
                var.validation_error = None

    except Exception as e:
        raise HTTPException(status_code=500, detail=str(e)) from e
    else:
        return filtered_variables


@router.patch("/{variable_id}", response_model=VariableRead, status_code=200, include_in_schema=False)
async def update_variable(
    *,
    session: DbSession,
    variable_id: UUID,
    variable: VariableUpdate,
    current_user: CurrentActiveUser,
):
    """Update a variable."""
    variable_service = get_variable_service()
    if not isinstance(variable_service, DatabaseVariableService):
        msg = "Variable service is not an instance of DatabaseVariableService"
        raise TypeError(msg)
    try:
        # Get existing variable to check if it's a model provider credential
        existing_variable = await variable_service.get_variable_by_id(
            user_id=current_user.id, variable_id=variable_id, session=session
        )

        # Validate API key if updating a model provider variable
        if existing_variable.name in model_provider_variable_mapping.values() and variable.value:
            provider = get_provider_from_variable_name(existing_variable.name)
            if provider is not None:
                # Run validation off the event loop to avoid blocking
                try:
                    await asyncio.to_thread(
                        validate_model_provider_key,
                        provider,
                        {existing_variable.name: variable.value},
                    )
                except ValueError as e:
                    raise HTTPException(status_code=400, detail=str(e)) from e

        return await variable_service.update_variable_fields(
            user_id=current_user.id,
            variable_id=variable_id,
            variable=variable,
            session=session,
        )
    except NoResultFound as e:
        raise HTTPException(status_code=404, detail="Variable not found") from e
    except ValueError as e:
        raise HTTPException(status_code=404, detail="Variable not found") from e
    except Exception as e:
        if isinstance(e, HTTPException):
            raise
        raise HTTPException(status_code=500, detail=str(e)) from e


@router.delete("/{variable_id}", status_code=204, include_in_schema=False)
async def delete_variable(
    *,
    session: DbSession,
    variable_id: UUID,
    current_user: CurrentActiveUser,
) -> None:
    """Delete a variable.

    If the deleted variable is a model provider credential (e.g., OPENAI_API_KEY),
    all disabled models for that provider are automatically cleared.
    """
    variable_service = get_variable_service()
    try:
        # Get the variable before deleting to check if it's a provider credential
        variable_to_delete = await variable_service.get_variable_by_id(
            user_id=current_user.id, variable_id=variable_id, session=session
        )

        # Check if this variable is a model provider credential
        provider = get_provider_from_variable_name(variable_to_delete.name)

        # Delete the variable
        await variable_service.delete_variable_by_id(user_id=current_user.id, variable_id=variable_id, session=session)

        # If this was a provider credential, clean up disabled and enabled models for that provider
        if provider and isinstance(variable_service, DatabaseVariableService):
            await _cleanup_provider_models(variable_service, current_user.id, provider, session)

    except Exception as e:
        raise HTTPException(status_code=500, detail=str(e)) from e


def _collect_candidate_variable_keys_from_flow_data(data: dict) -> set[str]:
    """Collect explicit global-variable keys from flow data."""
    candidate_keys: set[str] = set()

    for node in data.get("nodes", []):
        template = node.get("data", {}).get("node", {}).get("template", {})
        if not isinstance(template, dict):
            continue
        for field in template.values():
            if not isinstance(field, dict):
                continue
            if field.get("load_from_db") is True:
                var_name = field.get("value")
                normalized_var_name = var_name.strip() if isinstance(var_name, str) else None
                if normalized_var_name:
                    candidate_keys.add(normalized_var_name)

    return candidate_keys


def _validate_flow_or_422(*, version_id: UUID, data: object) -> dict:
    """Validate flow version data structure and raise HTTP 422 on malformed input."""
    if not (isinstance(data, dict) and "nodes" in data and isinstance(data["nodes"], list)):
        raise HTTPException(
            status_code=422,
            detail=(
                f"Flow version {version_id} data must be a JSON object with a 'nodes' list containing node templates."
            ),
        )
    return data


@router.post("/detections", response_model=DetectVarsResponse, include_in_schema=False)
async def detect_env_vars(
    payload: DetectVarsRequest,
    session: DbSession,
    current_user: CurrentActiveUser,
):
    """Detect global variable references used by the given flow version IDs.

    Candidates are inferred only from ``load_from_db=True`` fields, where
    the field value is interpreted as the referenced global variable name.

    Returned values are cross-checked against the user's existing global
    variable names. This is a security guardrail: it prevents echoing arbitrary
    template values (including accidental secrets) and ensures results are
    actual stored global variables.
    """
    variable_service = get_variable_service()
    existing_variable_names = {
        name
        for name in await variable_service.list_variables(user_id=current_user.id, session=session)
        if isinstance(name, str) and name
    }

    candidate_keys: set[str] = set()
    versions_by_id = await get_flow_version_entries_by_ids(
        session,
        version_ids=payload.flow_version_ids,
        user_id=current_user.id,
    )

    for version_id in payload.flow_version_ids:
        version = versions_by_id.get(version_id)
        if version is None:
            raise HTTPException(status_code=404, detail=f"Flow version {version_id} not found")

        data = _validate_flow_or_422(version_id=version_id, data=version.data)
        candidate_keys.update(_collect_candidate_variable_keys_from_flow_data(data))

    return DetectVarsResponse(variables=sorted(existing_variable_names.intersection(candidate_keys)))
