import json
import logging
import time as _time
import uuid
from collections.abc import Mapping
from typing import Any

from sqlalchemy import desc, func
from sqlalchemy.orm import Session

from configs import dify_config
from constants import HIDDEN_VALUE, UNKNOWN_VALUE
from core.helper.provider_cache import NoOpProviderCredentialCache
from core.helper.provider_encryption import ProviderConfigEncrypter, create_provider_encrypter
from core.plugin.entities.plugin_daemon import CredentialType
from core.plugin.impl.oauth import OAuthHandler
from core.tools.utils.system_oauth_encryption import decrypt_system_oauth_params
from core.trigger.entities.api_entities import (
    TriggerProviderApiEntity,
    TriggerProviderSubscriptionApiEntity,
)
from core.trigger.entities.entities import Subscription as TriggerSubscriptionEntity
from core.trigger.provider import PluginTriggerProviderController
from core.trigger.trigger_manager import TriggerManager
from core.trigger.utils.encryption import (
    create_trigger_provider_encrypter_for_properties,
    create_trigger_provider_encrypter_for_subscription,
    delete_cache_for_subscription,
)
from core.trigger.utils.endpoint import generate_plugin_trigger_endpoint_url
from extensions.ext_database import db
from extensions.ext_redis import redis_client
from models.provider_ids import TriggerProviderID
from models.trigger import (
    TriggerOAuthSystemClient,
    TriggerOAuthTenantClient,
    TriggerSubscription,
    WorkflowPluginTrigger,
)
from services.plugin.plugin_service import PluginService

logger = logging.getLogger(__name__)


class TriggerProviderService:
    """Service for managing trigger providers and credentials"""

    ##########################
    # Trigger provider
    ##########################
    __MAX_TRIGGER_PROVIDER_COUNT__ = 10

    @classmethod
    def get_trigger_provider(cls, tenant_id: str, provider: TriggerProviderID) -> TriggerProviderApiEntity:
        """Get info for a trigger provider"""
        return TriggerManager.get_trigger_provider(tenant_id, provider).to_api_entity()

    @classmethod
    def list_trigger_providers(cls, tenant_id: str) -> list[TriggerProviderApiEntity]:
        """List all trigger providers for the current tenant"""
        return [provider.to_api_entity() for provider in TriggerManager.list_all_trigger_providers(tenant_id)]

    @classmethod
    def list_trigger_provider_subscriptions(
        cls, tenant_id: str, provider_id: TriggerProviderID
    ) -> list[TriggerProviderSubscriptionApiEntity]:
        """List all trigger subscriptions for the current tenant"""
        subscriptions: list[TriggerProviderSubscriptionApiEntity] = []
        workflows_in_use_map: dict[str, int] = {}
        with Session(db.engine, expire_on_commit=False) as session:
            # Get all subscriptions
            subscriptions_db = (
                session.query(TriggerSubscription)
                .filter_by(tenant_id=tenant_id, provider_id=str(provider_id))
                .order_by(desc(TriggerSubscription.created_at))
                .all()
            )
            subscriptions = [subscription.to_api_entity() for subscription in subscriptions_db]
            if not subscriptions:
                return []
            usage_counts = (
                session.query(
                    WorkflowPluginTrigger.subscription_id,
                    func.count(func.distinct(WorkflowPluginTrigger.app_id)).label("app_count"),
                )
                .filter(
                    WorkflowPluginTrigger.tenant_id == tenant_id,
                    WorkflowPluginTrigger.subscription_id.in_([s.id for s in subscriptions]),
                )
                .group_by(WorkflowPluginTrigger.subscription_id)
                .all()
            )
            workflows_in_use_map = {str(row.subscription_id): int(row.app_count) for row in usage_counts}

        provider_controller = TriggerManager.get_trigger_provider(tenant_id, provider_id)
        for subscription in subscriptions:
            encrypter, _ = create_trigger_provider_encrypter_for_subscription(
                tenant_id=tenant_id,
                controller=provider_controller,
                subscription=subscription,
            )
            subscription.credentials = dict(
                encrypter.mask_credentials(dict(encrypter.decrypt(subscription.credentials)))
            )
            subscription.properties = dict(encrypter.mask_credentials(dict(encrypter.decrypt(subscription.properties))))
            subscription.parameters = dict(encrypter.mask_credentials(dict(encrypter.decrypt(subscription.parameters))))
            count = workflows_in_use_map.get(subscription.id)
            subscription.workflows_in_use = count if count is not None else 0

        return subscriptions

    @classmethod
    def add_trigger_subscription(
        cls,
        tenant_id: str,
        user_id: str,
        name: str,
        provider_id: TriggerProviderID,
        endpoint_id: str,
        credential_type: CredentialType,
        parameters: Mapping[str, Any],
        properties: Mapping[str, Any],
        credentials: Mapping[str, str],
        subscription_id: str | None = None,
        credential_expires_at: int = -1,
        expires_at: int = -1,
    ) -> Mapping[str, Any]:
        """
        Add a new trigger provider with credentials.
        Supports multiple credential instances per provider.

        :param tenant_id: Tenant ID
        :param provider_id: Provider identifier (e.g., "plugin_id/provider_name")
        :param credential_type: Type of credential (oauth or api_key)
        :param credentials: Credential data to encrypt and store
        :param name: Optional name for this credential instance
        :param expires_at: OAuth token expiration timestamp
        :return: Success response
        """
        try:
            provider_controller = TriggerManager.get_trigger_provider(tenant_id, provider_id)
            with Session(db.engine, expire_on_commit=False) as session:
                # Use distributed lock to prevent race conditions
                lock_key = f"trigger_provider_create_lock:{tenant_id}_{provider_id}"
                with redis_client.lock(lock_key, timeout=20):
                    # Check provider count limit
                    provider_count = (
                        session.query(TriggerSubscription)
                        .filter_by(tenant_id=tenant_id, provider_id=str(provider_id))
                        .count()
                    )

                    if provider_count >= cls.__MAX_TRIGGER_PROVIDER_COUNT__:
                        raise ValueError(
                            f"Maximum number of providers ({cls.__MAX_TRIGGER_PROVIDER_COUNT__}) "
                            f"reached for {provider_id}"
                        )

                    # Check if name already exists
                    existing = (
                        session.query(TriggerSubscription)
                        .filter_by(tenant_id=tenant_id, provider_id=str(provider_id), name=name)
                        .first()
                    )
                    if existing:
                        raise ValueError(f"Credential name '{name}' already exists for this provider")

                    credential_encrypter: ProviderConfigEncrypter | None = None
                    if credential_type != CredentialType.UNAUTHORIZED:
                        credential_encrypter, _ = create_provider_encrypter(
                            tenant_id=tenant_id,
                            config=provider_controller.get_credential_schema_config(credential_type),
                            cache=NoOpProviderCredentialCache(),
                        )

                    properties_encrypter, _ = create_provider_encrypter(
                        tenant_id=tenant_id,
                        config=provider_controller.get_properties_schema(),
                        cache=NoOpProviderCredentialCache(),
                    )

                    # Create provider record
                    subscription = TriggerSubscription(
                        tenant_id=tenant_id,
                        user_id=user_id,
                        name=name,
                        endpoint_id=endpoint_id,
                        provider_id=str(provider_id),
                        parameters=dict(parameters),
                        properties=dict(properties_encrypter.encrypt(dict(properties))),
                        credentials=dict(credential_encrypter.encrypt(dict(credentials)))
                        if credential_encrypter
                        else {},
                        credential_type=credential_type.value,
                        credential_expires_at=credential_expires_at,
                        expires_at=expires_at,
                    )
                    subscription.id = subscription_id or str(uuid.uuid4())

                    session.add(subscription)
                    session.commit()

                    return {
                        "result": "success",
                        "id": str(subscription.id),
                    }

        except Exception as e:
            logger.exception("Failed to add trigger provider")
            raise ValueError(str(e))

    @classmethod
    def get_subscription_by_id(cls, tenant_id: str, subscription_id: str | None = None) -> TriggerSubscription | None:
        """
        Get a trigger subscription by the ID.
        """
        with Session(db.engine, expire_on_commit=False) as session:
            subscription: TriggerSubscription | None = None
            if subscription_id:
                subscription = (
                    session.query(TriggerSubscription).filter_by(tenant_id=tenant_id, id=subscription_id).first()
                )
            else:
                subscription = session.query(TriggerSubscription).filter_by(tenant_id=tenant_id).first()
            if subscription:
                provider_controller = TriggerManager.get_trigger_provider(
                    tenant_id, TriggerProviderID(subscription.provider_id)
                )
                encrypter, _ = create_trigger_provider_encrypter_for_subscription(
                    tenant_id=tenant_id,
                    controller=provider_controller,
                    subscription=subscription,
                )
                subscription.credentials = dict(encrypter.decrypt(subscription.credentials))
                properties_encrypter, _ = create_trigger_provider_encrypter_for_properties(
                    tenant_id=subscription.tenant_id,
                    controller=provider_controller,
                    subscription=subscription,
                )
                subscription.properties = dict(properties_encrypter.decrypt(subscription.properties))
            return subscription

    @classmethod
    def delete_trigger_provider(cls, session: Session, tenant_id: str, subscription_id: str):
        """
        Delete a trigger provider subscription within an existing session.

        :param session: Database session
        :param tenant_id: Tenant ID
        :param subscription_id: Subscription instance ID
        :return: Success response
        """
        subscription: TriggerSubscription | None = (
            session.query(TriggerSubscription).filter_by(tenant_id=tenant_id, id=subscription_id).first()
        )
        if not subscription:
            raise ValueError(f"Trigger provider subscription {subscription_id} not found")

        credential_type: CredentialType = CredentialType.of(subscription.credential_type)
        is_auto_created: bool = credential_type in [CredentialType.OAUTH2, CredentialType.API_KEY]
        if is_auto_created:
            provider_id = TriggerProviderID(subscription.provider_id)
            provider_controller: PluginTriggerProviderController = TriggerManager.get_trigger_provider(
                tenant_id=tenant_id, provider_id=provider_id
            )
            encrypter, _ = create_trigger_provider_encrypter_for_subscription(
                tenant_id=tenant_id,
                controller=provider_controller,
                subscription=subscription,
            )
            try:
                TriggerManager.unsubscribe_trigger(
                    tenant_id=tenant_id,
                    user_id=subscription.user_id,
                    provider_id=provider_id,
                    subscription=subscription.to_entity(),
                    credentials=encrypter.decrypt(subscription.credentials),
                    credential_type=credential_type,
                )
            except Exception as e:
                logger.exception("Error unsubscribing trigger", exc_info=e)

        # Clear cache
        session.delete(subscription)
        delete_cache_for_subscription(
            tenant_id=tenant_id,
            provider_id=subscription.provider_id,
            subscription_id=subscription.id,
        )

    @classmethod
    def refresh_oauth_token(
        cls,
        tenant_id: str,
        subscription_id: str,
    ) -> Mapping[str, Any]:
        """
        Refresh OAuth token for a trigger provider.

        :param tenant_id: Tenant ID
        :param subscription_id: Subscription instance ID
        :return: New token info
        """
        with Session(db.engine) as session:
            subscription = session.query(TriggerSubscription).filter_by(tenant_id=tenant_id, id=subscription_id).first()

            if not subscription:
                raise ValueError(f"Trigger provider subscription {subscription_id} not found")

            if subscription.credential_type != CredentialType.OAUTH2.value:
                raise ValueError("Only OAuth credentials can be refreshed")

            provider_id = TriggerProviderID(subscription.provider_id)
            provider_controller: PluginTriggerProviderController = TriggerManager.get_trigger_provider(
                tenant_id=tenant_id, provider_id=provider_id
            )
            # Create encrypter
            encrypter, cache = create_provider_encrypter(
                tenant_id=tenant_id,
                config=[x.to_basic_provider_config() for x in provider_controller.get_oauth_client_schema()],
                cache=NoOpProviderCredentialCache(),
            )

            # Decrypt current credentials
            current_credentials = encrypter.decrypt(subscription.credentials)

            # Get OAuth client configuration
            redirect_uri = (
                f"{dify_config.CONSOLE_API_URL}/console/api/oauth/plugin/{subscription.provider_id}/trigger/callback"
            )
            system_credentials = cls.get_oauth_client(tenant_id, provider_id)

            # Refresh token
            oauth_handler = OAuthHandler()
            refreshed_credentials = oauth_handler.refresh_credentials(
                tenant_id=tenant_id,
                user_id=subscription.user_id,
                plugin_id=provider_id.plugin_id,
                provider=provider_id.provider_name,
                redirect_uri=redirect_uri,
                system_credentials=system_credentials or {},
                credentials=current_credentials,
            )

            # Update credentials
            subscription.credentials = dict(encrypter.encrypt(dict(refreshed_credentials.credentials)))
            subscription.credential_expires_at = refreshed_credentials.expires_at
            session.commit()

            # Clear cache
            cache.delete()

            return {
                "result": "success",
                "expires_at": refreshed_credentials.expires_at,
            }

    @classmethod
    def refresh_subscription(
        cls,
        tenant_id: str,
        subscription_id: str,
        now: int | None = None,
    ) -> Mapping[str, Any]:
        """
        Refresh trigger subscription if expired.

        Args:
            tenant_id: Tenant ID
            subscription_id: Subscription instance ID
            now: Current timestamp, defaults to `int(time.time())`

        Returns:
            Mapping with keys: `result` ("success"|"skipped") and `expires_at` (new or existing value)
        """
        now_ts: int = int(now if now is not None else _time.time())

        with Session(db.engine) as session:
            subscription: TriggerSubscription | None = (
                session.query(TriggerSubscription).filter_by(tenant_id=tenant_id, id=subscription_id).first()
            )
            if subscription is None:
                raise ValueError(f"Trigger provider subscription {subscription_id} not found")

            if subscription.expires_at == -1 or int(subscription.expires_at) > now_ts:
                logger.debug(
                    "Subscription not due for refresh: tenant=%s id=%s expires_at=%s now=%s",
                    tenant_id,
                    subscription_id,
                    subscription.expires_at,
                    now_ts,
                )
                return {"result": "skipped", "expires_at": int(subscription.expires_at)}

            provider_id = TriggerProviderID(subscription.provider_id)
            controller: PluginTriggerProviderController = TriggerManager.get_trigger_provider(
                tenant_id=tenant_id, provider_id=provider_id
            )

            # Decrypt credentials and properties for runtime
            credential_encrypter, _ = create_trigger_provider_encrypter_for_subscription(
                tenant_id=tenant_id,
                controller=controller,
                subscription=subscription,
            )
            properties_encrypter, properties_cache = create_trigger_provider_encrypter_for_properties(
                tenant_id=tenant_id,
                controller=controller,
                subscription=subscription,
            )

            decrypted_credentials = credential_encrypter.decrypt(subscription.credentials)
            decrypted_properties = properties_encrypter.decrypt(subscription.properties)

            sub_entity: TriggerSubscriptionEntity = TriggerSubscriptionEntity(
                expires_at=int(subscription.expires_at),
                endpoint=generate_plugin_trigger_endpoint_url(subscription.endpoint_id),
                parameters=subscription.parameters,
                properties=decrypted_properties,
            )

            refreshed: TriggerSubscriptionEntity = controller.refresh_trigger(
                subscription=sub_entity,
                credentials=decrypted_credentials,
                credential_type=CredentialType.of(subscription.credential_type),
            )

            # Persist refreshed properties and expires_at
            subscription.properties = dict(properties_encrypter.encrypt(dict(refreshed.properties)))
            subscription.expires_at = int(refreshed.expires_at)
            session.commit()
            properties_cache.delete()

            logger.info(
                "Subscription refreshed (service): tenant=%s id=%s new_expires_at=%s",
                tenant_id,
                subscription_id,
                subscription.expires_at,
            )

            return {"result": "success", "expires_at": int(refreshed.expires_at)}

    @classmethod
    def get_oauth_client(cls, tenant_id: str, provider_id: TriggerProviderID) -> Mapping[str, Any] | None:
        """
        Get OAuth client configuration for a provider.
        First tries tenant-level OAuth, then falls back to system OAuth.

        :param tenant_id: Tenant ID
        :param provider_id: Provider identifier
        :return: OAuth client configuration or None
        """
        provider_controller: PluginTriggerProviderController = TriggerManager.get_trigger_provider(
            tenant_id=tenant_id, provider_id=provider_id
        )
        with Session(db.engine, expire_on_commit=False) as session:
            tenant_client: TriggerOAuthTenantClient | None = (
                session.query(TriggerOAuthTenantClient)
                .filter_by(
                    tenant_id=tenant_id,
                    provider=provider_id.provider_name,
                    plugin_id=provider_id.plugin_id,
                    enabled=True,
                )
                .first()
            )

            oauth_params: Mapping[str, Any] | None = None
            if tenant_client:
                encrypter, _ = create_provider_encrypter(
                    tenant_id=tenant_id,
                    config=[x.to_basic_provider_config() for x in provider_controller.get_oauth_client_schema()],
                    cache=NoOpProviderCredentialCache(),
                )
                oauth_params = encrypter.decrypt(dict(tenant_client.oauth_params))
                return oauth_params

            is_verified = PluginService.is_plugin_verified(tenant_id, provider_controller.plugin_unique_identifier)
            if not is_verified:
                return None

            # Check for system-level OAuth client
            system_client: TriggerOAuthSystemClient | None = (
                session.query(TriggerOAuthSystemClient)
                .filter_by(plugin_id=provider_id.plugin_id, provider=provider_id.provider_name)
                .first()
            )

            if system_client:
                try:
                    oauth_params = decrypt_system_oauth_params(system_client.encrypted_oauth_params)
                except Exception as e:
                    raise ValueError(f"Error decrypting system oauth params: {e}")

            return oauth_params

    @classmethod
    def is_oauth_system_client_exists(cls, tenant_id: str, provider_id: TriggerProviderID) -> bool:
        """
        Check if system OAuth client exists for a trigger provider.
        """
        provider_controller = TriggerManager.get_trigger_provider(tenant_id=tenant_id, provider_id=provider_id)
        is_verified = PluginService.is_plugin_verified(tenant_id, provider_controller.plugin_unique_identifier)
        if not is_verified:
            return False
        with Session(db.engine, expire_on_commit=False) as session:
            system_client: TriggerOAuthSystemClient | None = (
                session.query(TriggerOAuthSystemClient)
                .filter_by(plugin_id=provider_id.plugin_id, provider=provider_id.provider_name)
                .first()
            )
            return system_client is not None

    @classmethod
    def save_custom_oauth_client_params(
        cls,
        tenant_id: str,
        provider_id: TriggerProviderID,
        client_params: Mapping[str, Any] | None = None,
        enabled: bool | None = None,
    ) -> Mapping[str, Any]:
        """
        Save or update custom OAuth client parameters for a trigger provider.

        :param tenant_id: Tenant ID
        :param provider_id: Provider identifier
        :param client_params: OAuth client parameters (client_id, client_secret, etc.)
        :param enabled: Enable/disable the custom OAuth client
        :return: Success response
        """
        if client_params is None and enabled is None:
            return {"result": "success"}

        # Get provider controller to access schema
        provider_controller: PluginTriggerProviderController = TriggerManager.get_trigger_provider(
            tenant_id=tenant_id, provider_id=provider_id
        )

        with Session(db.engine) as session:
            # Find existing custom client params
            custom_client = (
                session.query(TriggerOAuthTenantClient)
                .filter_by(
                    tenant_id=tenant_id,
                    plugin_id=provider_id.plugin_id,
                    provider=provider_id.provider_name,
                )
                .first()
            )

            # Create new record if doesn't exist
            if custom_client is None:
                custom_client = TriggerOAuthTenantClient(
                    tenant_id=tenant_id,
                    plugin_id=provider_id.plugin_id,
                    provider=provider_id.provider_name,
                )
                session.add(custom_client)

            # Update client params if provided
            if client_params is None:
                custom_client.encrypted_oauth_params = json.dumps({})
            else:
                encrypter, cache = create_provider_encrypter(
                    tenant_id=tenant_id,
                    config=[x.to_basic_provider_config() for x in provider_controller.get_oauth_client_schema()],
                    cache=NoOpProviderCredentialCache(),
                )

                # Handle hidden values
                original_params = encrypter.decrypt(dict(custom_client.oauth_params))
                new_params: dict[str, Any] = {
                    key: value if value != HIDDEN_VALUE else original_params.get(key, UNKNOWN_VALUE)
                    for key, value in client_params.items()
                }
                custom_client.encrypted_oauth_params = json.dumps(encrypter.encrypt(new_params))
                cache.delete()

            # Update enabled status if provided
            if enabled is not None:
                custom_client.enabled = enabled

            session.commit()

        return {"result": "success"}

    @classmethod
    def get_custom_oauth_client_params(cls, tenant_id: str, provider_id: TriggerProviderID) -> Mapping[str, Any]:
        """
        Get custom OAuth client parameters for a trigger provider.

        :param tenant_id: Tenant ID
        :param provider_id: Provider identifier
        :return: Masked OAuth client parameters
        """
        with Session(db.engine) as session:
            custom_client = (
                session.query(TriggerOAuthTenantClient)
                .filter_by(
                    tenant_id=tenant_id,
                    plugin_id=provider_id.plugin_id,
                    provider=provider_id.provider_name,
                )
                .first()
            )

            if custom_client is None:
                return {}

            # Get provider controller to access schema
            provider_controller: PluginTriggerProviderController = TriggerManager.get_trigger_provider(
                tenant_id=tenant_id, provider_id=provider_id
            )

            # Create encrypter to decrypt and mask values
            encrypter, _ = create_provider_encrypter(
                tenant_id=tenant_id,
                config=[x.to_basic_provider_config() for x in provider_controller.get_oauth_client_schema()],
                cache=NoOpProviderCredentialCache(),
            )

            return encrypter.mask_plugin_credentials(encrypter.decrypt(dict(custom_client.oauth_params)))

    @classmethod
    def delete_custom_oauth_client_params(cls, tenant_id: str, provider_id: TriggerProviderID) -> Mapping[str, Any]:
        """
        Delete custom OAuth client parameters for a trigger provider.

        :param tenant_id: Tenant ID
        :param provider_id: Provider identifier
        :return: Success response
        """
        with Session(db.engine) as session:
            session.query(TriggerOAuthTenantClient).filter_by(
                tenant_id=tenant_id,
                provider=provider_id.provider_name,
                plugin_id=provider_id.plugin_id,
            ).delete()
            session.commit()

        return {"result": "success"}

    @classmethod
    def is_oauth_custom_client_enabled(cls, tenant_id: str, provider_id: TriggerProviderID) -> bool:
        """
        Check if custom OAuth client is enabled for a trigger provider.

        :param tenant_id: Tenant ID
        :param provider_id: Provider identifier
        :return: True if enabled, False otherwise
        """
        with Session(db.engine, expire_on_commit=False) as session:
            custom_client = (
                session.query(TriggerOAuthTenantClient)
                .filter_by(
                    tenant_id=tenant_id,
                    plugin_id=provider_id.plugin_id,
                    provider=provider_id.provider_name,
                    enabled=True,
                )
                .first()
            )
            return custom_client is not None

    @classmethod
    def get_subscription_by_endpoint(cls, endpoint_id: str) -> TriggerSubscription | None:
        """
        Get a trigger subscription by the endpoint ID.
        """
        with Session(db.engine, expire_on_commit=False) as session:
            subscription = session.query(TriggerSubscription).filter_by(endpoint_id=endpoint_id).first()
            if not subscription:
                return None
            provider_controller: PluginTriggerProviderController = TriggerManager.get_trigger_provider(
                tenant_id=subscription.tenant_id, provider_id=TriggerProviderID(subscription.provider_id)
            )
            credential_encrypter, _ = create_trigger_provider_encrypter_for_subscription(
                tenant_id=subscription.tenant_id,
                controller=provider_controller,
                subscription=subscription,
            )
            subscription.credentials = dict(credential_encrypter.decrypt(subscription.credentials))

            properties_encrypter, _ = create_trigger_provider_encrypter_for_properties(
                tenant_id=subscription.tenant_id,
                controller=provider_controller,
                subscription=subscription,
            )
            subscription.properties = dict(properties_encrypter.decrypt(subscription.properties))
            return subscription