import logging
import time
from collections.abc import Mapping
from typing import TYPE_CHECKING, Any

if TYPE_CHECKING:
    from models.account import Account

from sqlalchemy import delete, func, select, update
from sqlalchemy.orm import Session, sessionmaker

from configs import dify_config
from constants import HIDDEN_VALUE, UNKNOWN_VALUE
from core.helper import encrypter
from core.helper.name_generator import generate_incremental_name
from core.helper.provider_cache import NoOpProviderCredentialCache
from core.plugin.entities.plugin_daemon import CredentialType
from core.plugin.impl.datasource import PluginDatasourceManager
from core.plugin.impl.oauth import OAuthHandler
from core.plugin.plugin_service import PluginService
from core.tools.utils.encryption import ProviderConfigCache, ProviderConfigEncrypter, create_provider_encrypter
from extensions.ext_database import db
from extensions.ext_redis import redis_client
from graphon.model_runtime.entities.provider_entities import FormType
from models.oauth import DatasourceOauthParamConfig, DatasourceOauthTenantParamConfig, DatasourceProvider
from models.provider_ids import DatasourceProviderID

logger = logging.getLogger(__name__)


def get_current_user():
    from libs.login import current_user
    from models.account import Account
    from models.model import EndUser

    try:
        user_object = current_user._get_current_object()
    except AttributeError:
        # Handle case where current_user might not be a LocalProxy in test environments
        user_object = current_user

    if not isinstance(user_object, (Account, EndUser)):
        raise TypeError(f"current_user must be Account or EndUser, got {type(user_object).__name__}")
    return current_user


class DatasourceProviderService:
    """
    Model Provider Service
    """

    def __init__(self) -> None:
        self.provider_manager = PluginDatasourceManager()

    @staticmethod
    def _should_refresh_credentials(datasource_provider: DatasourceProvider, now: int | None = None) -> bool:
        current_time = int(time.time()) if now is None else now
        if datasource_provider.expires_at == -1:
            return False
        return (datasource_provider.expires_at - 60) < current_time

    def _refresh_datasource_credentials(
        self,
        tenant_id: str,
        provider: str,
        plugin_id: str,
        datasource_provider: DatasourceProvider,
        current_user: Any,
    ) -> tuple[dict[str, Any], int]:
        datasource_provider_id = DatasourceProviderID(f"{plugin_id}/{provider}")
        provider_name = datasource_provider_id.provider_name
        credential_id = getattr(datasource_provider, "id", None)
        credential_name = getattr(datasource_provider, "name", None)
        logger.info(
            "Refreshing datasource credentials for provider %s",
            provider_name,
            extra={
                "tenant_id": tenant_id,
                "plugin_id": datasource_provider_id.plugin_id,
                "provider": provider_name,
                "credential_id": credential_id,
                "credential_name": credential_name,
                "expires_at": datasource_provider.expires_at,
            },
        )
        decrypted_credentials = self.decrypt_datasource_provider_credentials(
            tenant_id=tenant_id,
            datasource_provider=datasource_provider,
            plugin_id=plugin_id,
            provider=provider,
        )
        redirect_uri = (
            f"{dify_config.CONSOLE_API_URL}/console/api/oauth/plugin/{datasource_provider_id}/datasource/callback"
        )
        system_credentials = self.get_oauth_client(tenant_id, datasource_provider_id)
        try:
            refreshed_credentials = OAuthHandler().refresh_credentials(
                tenant_id=tenant_id,
                user_id=current_user.id,
                plugin_id=datasource_provider_id.plugin_id,
                provider=provider_name,
                redirect_uri=redirect_uri,
                system_credentials=system_credentials or {},
                credentials=decrypted_credentials,
            )
        except Exception as exc:
            message = (
                f"Failed to refresh datasource credentials for provider {provider_name}"
                f" (credential: {credential_name or credential_id or 'unknown'})"
            )
            logger.exception(
                message,
                extra={
                    "tenant_id": tenant_id,
                    "plugin_id": datasource_provider_id.plugin_id,
                    "provider": provider_name,
                    "credential_id": credential_id,
                    "credential_name": credential_name,
                },
            )
            raise ValueError(f"{message}: {exc}") from exc
        encrypted_credentials = self.encrypt_datasource_provider_credentials(
            tenant_id=tenant_id,
            raw_credentials=refreshed_credentials.credentials,
            provider=provider,
            plugin_id=plugin_id,
            datasource_provider=datasource_provider,
        )
        logger.info(
            "Refreshed datasource credentials for provider %s",
            provider_name,
            extra={
                "tenant_id": tenant_id,
                "plugin_id": datasource_provider_id.plugin_id,
                "provider": provider_name,
                "credential_id": credential_id,
                "credential_name": credential_name,
                "expires_at": refreshed_credentials.expires_at,
            },
        )
        return encrypted_credentials, refreshed_credentials.expires_at

    def remove_oauth_custom_client_params(self, tenant_id: str, datasource_provider_id: DatasourceProviderID):
        """
        remove oauth custom client params
        """
        with sessionmaker(bind=db.engine).begin() as session:
            session.execute(
                delete(DatasourceOauthTenantParamConfig).where(
                    DatasourceOauthTenantParamConfig.tenant_id == tenant_id,
                    DatasourceOauthTenantParamConfig.provider == datasource_provider_id.provider_name,
                    DatasourceOauthTenantParamConfig.plugin_id == datasource_provider_id.plugin_id,
                )
            )

    def decrypt_datasource_provider_credentials(
        self,
        tenant_id: str,
        datasource_provider: DatasourceProvider,
        plugin_id: str,
        provider: str,
    ) -> dict[str, Any]:
        encrypted_credentials = datasource_provider.encrypted_credentials
        credential_secret_variables = self.extract_secret_variables(
            tenant_id=tenant_id,
            provider_id=f"{plugin_id}/{provider}",
            credential_type=CredentialType.of(datasource_provider.auth_type),
        )
        decrypted_credentials = encrypted_credentials.copy()
        for key, value in decrypted_credentials.items():
            if key in credential_secret_variables:
                decrypted_credentials[key] = encrypter.decrypt_token(tenant_id, value)
        return decrypted_credentials

    def encrypt_datasource_provider_credentials(
        self,
        tenant_id: str,
        provider: str,
        plugin_id: str,
        raw_credentials: Mapping[str, Any],
        datasource_provider: DatasourceProvider,
    ) -> dict[str, Any]:
        provider_credential_secret_variables = self.extract_secret_variables(
            tenant_id=tenant_id,
            provider_id=f"{plugin_id}/{provider}",
            credential_type=CredentialType.of(datasource_provider.auth_type),
        )
        encrypted_credentials = dict(raw_credentials)
        for key, value in encrypted_credentials.items():
            if key in provider_credential_secret_variables:
                encrypted_credentials[key] = encrypter.encrypt_token(tenant_id, value)
        return encrypted_credentials

    def get_datasource_credentials(
        self,
        tenant_id: str,
        provider: str,
        plugin_id: str,
        credential_id: str | None = None,
    ) -> dict[str, Any]:
        """
        Return decrypted datasource credentials.

        If the stored credential is expired or about to expire, this method refreshes
        it through plugin-daemon and persists the refreshed credential before returning.
        """
        with sessionmaker(bind=db.engine).begin() as session:
            if credential_id:
                datasource_provider = session.scalar(
                    select(DatasourceProvider)
                    .where(DatasourceProvider.tenant_id == tenant_id, DatasourceProvider.id == credential_id)
                    .limit(1)
                )
            else:
                datasource_provider = session.scalar(
                    select(DatasourceProvider)
                    .where(
                        DatasourceProvider.tenant_id == tenant_id,
                        DatasourceProvider.provider == provider,
                        DatasourceProvider.plugin_id == plugin_id,
                    )
                    .order_by(DatasourceProvider.is_default.desc(), DatasourceProvider.created_at.asc())
                    .limit(1)
                )
            if not datasource_provider:
                return {}
            if self._should_refresh_credentials(datasource_provider):
                current_user = get_current_user()
                encrypted_credentials, expires_at = self._refresh_datasource_credentials(
                    tenant_id=tenant_id,
                    provider=provider,
                    plugin_id=plugin_id,
                    datasource_provider=datasource_provider,
                    current_user=current_user,
                )
                datasource_provider.encrypted_credentials = encrypted_credentials
                datasource_provider.expires_at = expires_at

            return self.decrypt_datasource_provider_credentials(
                tenant_id=tenant_id,
                datasource_provider=datasource_provider,
                plugin_id=plugin_id,
                provider=provider,
            )

    def get_all_datasource_credentials_by_provider(
        self,
        tenant_id: str,
        provider: str,
        plugin_id: str,
    ) -> list[dict[str, Any]]:
        """
        Return all decrypted datasource credentials for a provider.

        Expired credentials are refreshed independently. A failed credential refresh is
        logged and skipped so one broken authorization does not block other credentials.
        """
        with sessionmaker(bind=db.engine).begin() as session:
            datasource_providers = session.scalars(
                select(DatasourceProvider)
                .where(
                    DatasourceProvider.tenant_id == tenant_id,
                    DatasourceProvider.provider == provider,
                    DatasourceProvider.plugin_id == plugin_id,
                )
                .order_by(DatasourceProvider.is_default.desc(), DatasourceProvider.created_at.asc())
            ).all()
            if not datasource_providers:
                return []
            current_user = get_current_user()
            real_credentials_list = []
            for datasource_provider in datasource_providers:
                try:
                    if self._should_refresh_credentials(datasource_provider):
                        encrypted_credentials, expires_at = self._refresh_datasource_credentials(
                            tenant_id=tenant_id,
                            provider=provider,
                            plugin_id=plugin_id,
                            datasource_provider=datasource_provider,
                            current_user=current_user,
                        )
                        datasource_provider.encrypted_credentials = encrypted_credentials
                        datasource_provider.expires_at = expires_at
                    real_credentials = self.decrypt_datasource_provider_credentials(
                        tenant_id=tenant_id,
                        datasource_provider=datasource_provider,
                        plugin_id=plugin_id,
                        provider=provider,
                    )
                    real_credentials_list.append(real_credentials)
                except Exception:
                    logger.exception(
                        "Skipping datasource credentials for provider %s after refresh or decrypt failure",
                        provider,
                        extra={
                            "tenant_id": tenant_id,
                            "plugin_id": plugin_id,
                            "provider": provider,
                            "credential_id": getattr(datasource_provider, "id", None),
                            "credential_name": getattr(datasource_provider, "name", None),
                            "expires_at": getattr(datasource_provider, "expires_at", None),
                        },
                    )

            return real_credentials_list

    def update_datasource_provider_name(
        self, tenant_id: str, datasource_provider_id: DatasourceProviderID, name: str, credential_id: str
    ):
        """
        update datasource provider name
        """
        with sessionmaker(bind=db.engine).begin() as session:
            target_provider = session.scalar(
                select(DatasourceProvider)
                .where(
                    DatasourceProvider.tenant_id == tenant_id,
                    DatasourceProvider.id == credential_id,
                    DatasourceProvider.provider == datasource_provider_id.provider_name,
                    DatasourceProvider.plugin_id == datasource_provider_id.plugin_id,
                )
                .limit(1)
            )
            if target_provider is None:
                raise ValueError("provider not found")

            if target_provider.name == name:
                return

            # check name is exist
            if (
                session.scalar(
                    select(func.count(DatasourceProvider.id)).where(
                        DatasourceProvider.tenant_id == tenant_id,
                        DatasourceProvider.name == name,
                        DatasourceProvider.provider == datasource_provider_id.provider_name,
                        DatasourceProvider.plugin_id == datasource_provider_id.plugin_id,
                    )
                )
                or 0
            ) > 0:
                raise ValueError("Authorization name is already exists")

            target_provider.name = name
        return

    def set_default_datasource_provider(
        self, tenant_id: str, datasource_provider_id: DatasourceProviderID, credential_id: str
    ):
        """
        set default datasource provider
        """
        with sessionmaker(bind=db.engine).begin() as session:
            # get provider
            target_provider = session.scalar(
                select(DatasourceProvider)
                .where(
                    DatasourceProvider.tenant_id == tenant_id,
                    DatasourceProvider.id == credential_id,
                    DatasourceProvider.provider == datasource_provider_id.provider_name,
                    DatasourceProvider.plugin_id == datasource_provider_id.plugin_id,
                )
                .limit(1)
            )
            if target_provider is None:
                raise ValueError("provider not found")

            # clear default provider
            session.execute(
                update(DatasourceProvider)
                .where(
                    DatasourceProvider.tenant_id == tenant_id,
                    DatasourceProvider.provider == target_provider.provider,
                    DatasourceProvider.plugin_id == target_provider.plugin_id,
                    DatasourceProvider.is_default.is_(True),
                )
                .values(is_default=False)
                .execution_options(synchronize_session=False)
            )

            # set new default provider
            target_provider.is_default = True
        return {"result": "success"}

    def setup_oauth_custom_client_params(
        self,
        tenant_id: str,
        datasource_provider_id: DatasourceProviderID,
        client_params: dict[str, Any] | None,
        enabled: bool | None,
    ):
        """
        setup oauth custom client params
        """
        if client_params is None and enabled is None:
            return
        with sessionmaker(bind=db.engine).begin() as session:
            tenant_oauth_client_params = session.scalar(
                select(DatasourceOauthTenantParamConfig)
                .where(
                    DatasourceOauthTenantParamConfig.tenant_id == tenant_id,
                    DatasourceOauthTenantParamConfig.provider == datasource_provider_id.provider_name,
                    DatasourceOauthTenantParamConfig.plugin_id == datasource_provider_id.plugin_id,
                )
                .limit(1)
            )

            if not tenant_oauth_client_params:
                tenant_oauth_client_params = DatasourceOauthTenantParamConfig(
                    tenant_id=tenant_id,
                    provider=datasource_provider_id.provider_name,
                    plugin_id=datasource_provider_id.plugin_id,
                    client_params={},
                    enabled=False,
                )
                session.add(tenant_oauth_client_params)

            if client_params is not None:
                encrypter, _ = self.get_oauth_encrypter(tenant_id, datasource_provider_id)
                original_params = (
                    encrypter.decrypt(tenant_oauth_client_params.client_params) if tenant_oauth_client_params else {}
                )
                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()
                }
                tenant_oauth_client_params.client_params = dict(encrypter.encrypt(new_params))

            if enabled is not None:
                tenant_oauth_client_params.enabled = enabled

    def is_system_oauth_params_exist(self, datasource_provider_id: DatasourceProviderID) -> bool:
        """
        check if system oauth params exist
        """
        with Session(db.engine).no_autoflush as session:
            return (
                session.scalar(
                    select(DatasourceOauthParamConfig)
                    .where(
                        DatasourceOauthParamConfig.provider == datasource_provider_id.provider_name,
                        DatasourceOauthParamConfig.plugin_id == datasource_provider_id.plugin_id,
                    )
                    .limit(1)
                )
                is not None
            )

    def is_tenant_oauth_params_enabled(self, tenant_id: str, datasource_provider_id: DatasourceProviderID) -> bool:
        """
        check if tenant oauth params is enabled
        """
        return (
            db.session.scalar(
                select(func.count(DatasourceOauthTenantParamConfig.id)).where(
                    DatasourceOauthTenantParamConfig.tenant_id == tenant_id,
                    DatasourceOauthTenantParamConfig.provider == datasource_provider_id.provider_name,
                    DatasourceOauthTenantParamConfig.plugin_id == datasource_provider_id.plugin_id,
                    DatasourceOauthTenantParamConfig.enabled == True,
                )
            )
            or 0
        ) > 0

    def get_tenant_oauth_client(
        self, tenant_id: str, datasource_provider_id: DatasourceProviderID, mask: bool = False
    ) -> Mapping[str, Any] | None:
        """
        get tenant oauth client
        """
        tenant_oauth_client_params = db.session.scalar(
            select(DatasourceOauthTenantParamConfig)
            .where(
                DatasourceOauthTenantParamConfig.tenant_id == tenant_id,
                DatasourceOauthTenantParamConfig.provider == datasource_provider_id.provider_name,
                DatasourceOauthTenantParamConfig.plugin_id == datasource_provider_id.plugin_id,
            )
            .limit(1)
        )
        if tenant_oauth_client_params:
            encrypter, _ = self.get_oauth_encrypter(tenant_id, datasource_provider_id)
            if mask:
                return encrypter.mask_plugin_credentials(encrypter.decrypt(tenant_oauth_client_params.client_params))
            else:
                return encrypter.decrypt(tenant_oauth_client_params.client_params)
        return None

    def get_oauth_encrypter(
        self, tenant_id: str, datasource_provider_id: DatasourceProviderID
    ) -> tuple[ProviderConfigEncrypter, ProviderConfigCache]:
        """
        get oauth encrypter
        """
        datasource_provider = self.provider_manager.fetch_datasource_provider(
            tenant_id=tenant_id, provider_id=str(datasource_provider_id)
        )
        if not datasource_provider.declaration.oauth_schema:
            raise ValueError("Datasource provider oauth schema not found")

        client_schema = datasource_provider.declaration.oauth_schema.client_schema
        return create_provider_encrypter(
            tenant_id=tenant_id,
            config=[x.to_basic_provider_config() for x in client_schema],
            cache=NoOpProviderCredentialCache(),
        )

    def get_oauth_client(self, tenant_id: str, datasource_provider_id: DatasourceProviderID) -> dict[str, Any] | None:
        """
        get oauth client
        """
        provider = datasource_provider_id.provider_name
        plugin_id = datasource_provider_id.plugin_id
        with Session(db.engine).no_autoflush as session:
            # get tenant oauth client params
            tenant_oauth_client_params = session.scalar(
                select(DatasourceOauthTenantParamConfig)
                .where(
                    DatasourceOauthTenantParamConfig.tenant_id == tenant_id,
                    DatasourceOauthTenantParamConfig.provider == provider,
                    DatasourceOauthTenantParamConfig.plugin_id == plugin_id,
                    DatasourceOauthTenantParamConfig.enabled.is_(True),
                )
                .limit(1)
            )
            if tenant_oauth_client_params:
                encrypter, _ = self.get_oauth_encrypter(tenant_id, datasource_provider_id)
                return dict(encrypter.decrypt(tenant_oauth_client_params.client_params))

            provider_controller = self.provider_manager.fetch_datasource_provider(
                tenant_id=tenant_id, provider_id=str(datasource_provider_id)
            )
            is_verified = PluginService.is_plugin_verified(tenant_id, provider_controller.plugin_unique_identifier)
            if is_verified:
                # fallback to system oauth client params
                oauth_client_params = session.scalar(
                    select(DatasourceOauthParamConfig)
                    .where(
                        DatasourceOauthParamConfig.provider == provider,
                        DatasourceOauthParamConfig.plugin_id == plugin_id,
                    )
                    .limit(1)
                )
                if oauth_client_params:
                    return oauth_client_params.system_credentials

            raise ValueError(f"Please configure oauth client params(system/tenant) for {plugin_id}/{provider}")

    @staticmethod
    def generate_next_datasource_provider_name(
        session: Session, tenant_id: str, provider_id: DatasourceProviderID, credential_type: CredentialType
    ) -> str:
        db_providers = session.scalars(
            select(DatasourceProvider).where(
                DatasourceProvider.tenant_id == tenant_id,
                DatasourceProvider.provider == provider_id.provider_name,
                DatasourceProvider.plugin_id == provider_id.plugin_id,
            )
        ).all()
        return generate_incremental_name(
            [provider.name for provider in db_providers],
            f"{credential_type.get_name()}",
        )

    def reauthorize_datasource_oauth_provider(
        self,
        name: str | None,
        tenant_id: str,
        provider_id: DatasourceProviderID,
        avatar_url: str | None,
        expire_at: int,
        credentials: dict[str, Any],
        credential_id: str,
    ) -> None:
        """
        update datasource oauth provider
        """
        with sessionmaker(bind=db.engine).begin() as session:
            lock = f"datasource_provider_create_lock:{tenant_id}_{provider_id}_{CredentialType.OAUTH2.value}"
            with redis_client.lock(lock, timeout=20):
                target_provider = session.scalar(
                    select(DatasourceProvider)
                    .where(DatasourceProvider.id == credential_id, DatasourceProvider.tenant_id == tenant_id)
                    .limit(1)
                )
                if target_provider is None:
                    raise ValueError("provider not found")

                db_provider_name = name
                if not db_provider_name:
                    db_provider_name = target_provider.name
                else:
                    name_conflict = (
                        session.scalar(
                            select(func.count(DatasourceProvider.id)).where(
                                DatasourceProvider.tenant_id == tenant_id,
                                DatasourceProvider.name == db_provider_name,
                                DatasourceProvider.provider == provider_id.provider_name,
                                DatasourceProvider.plugin_id == provider_id.plugin_id,
                                DatasourceProvider.auth_type == CredentialType.OAUTH2.value,
                            )
                        )
                        or 0
                    )
                    if name_conflict > 0:
                        db_provider_name = generate_incremental_name(
                            [
                                provider.name
                                for provider in session.scalars(
                                    select(DatasourceProvider).where(
                                        DatasourceProvider.tenant_id == tenant_id,
                                        DatasourceProvider.provider == provider_id.provider_name,
                                        DatasourceProvider.plugin_id == provider_id.plugin_id,
                                    )
                                ).all()
                            ],
                            db_provider_name,
                        )

                provider_credential_secret_variables = self.extract_secret_variables(
                    tenant_id=tenant_id, provider_id=f"{provider_id}", credential_type=CredentialType.OAUTH2
                )
                for key, value in credentials.items():
                    if key in provider_credential_secret_variables:
                        credentials[key] = encrypter.encrypt_token(tenant_id, value)

                target_provider.expires_at = expire_at
                target_provider.encrypted_credentials = credentials
                target_provider.avatar_url = avatar_url or target_provider.avatar_url

    def add_datasource_oauth_provider(
        self,
        name: str | None,
        tenant_id: str,
        provider_id: DatasourceProviderID,
        avatar_url: str | None,
        expire_at: int,
        credentials: dict[str, Any],
    ) -> None:
        """
        add datasource oauth provider
        """
        credential_type = CredentialType.OAUTH2
        with sessionmaker(bind=db.engine).begin() as session:
            lock = f"datasource_provider_create_lock:{tenant_id}_{provider_id}_{credential_type.value}"
            with redis_client.lock(lock, timeout=60):
                db_provider_name = name
                if not db_provider_name:
                    db_provider_name = self.generate_next_datasource_provider_name(
                        session=session,
                        tenant_id=tenant_id,
                        provider_id=provider_id,
                        credential_type=credential_type,
                    )
                else:
                    if (
                        session.scalar(
                            select(func.count(DatasourceProvider.id)).where(
                                DatasourceProvider.tenant_id == tenant_id,
                                DatasourceProvider.name == db_provider_name,
                                DatasourceProvider.provider == provider_id.provider_name,
                                DatasourceProvider.plugin_id == provider_id.plugin_id,
                                DatasourceProvider.auth_type == credential_type.value,
                            )
                        )
                        or 0
                    ) > 0:
                        db_provider_name = generate_incremental_name(
                            [
                                provider.name
                                for provider in session.scalars(
                                    select(DatasourceProvider).where(
                                        DatasourceProvider.tenant_id == tenant_id,
                                        DatasourceProvider.provider == provider_id.provider_name,
                                        DatasourceProvider.plugin_id == provider_id.plugin_id,
                                    )
                                ).all()
                            ],
                            db_provider_name,
                        )

                provider_credential_secret_variables = self.extract_secret_variables(
                    tenant_id=tenant_id, provider_id=f"{provider_id}", credential_type=credential_type
                )
                for key, value in credentials.items():
                    if key in provider_credential_secret_variables:
                        credentials[key] = encrypter.encrypt_token(tenant_id, value)

                datasource_provider = DatasourceProvider(
                    tenant_id=tenant_id,
                    name=db_provider_name,
                    provider=provider_id.provider_name,
                    plugin_id=provider_id.plugin_id,
                    auth_type=credential_type.value,
                    encrypted_credentials=credentials,
                    avatar_url=avatar_url or "default",
                    expires_at=expire_at,
                )
                session.add(datasource_provider)

    def add_datasource_api_key_provider(
        self,
        name: str | None,
        tenant_id: str,
        provider_id: DatasourceProviderID,
        credentials: dict[str, Any],
    ) -> None:
        """
        validate datasource provider credentials.

        :param tenant_id:
        :param provider:
        :param credentials:
        """
        provider_name = provider_id.provider_name
        plugin_id = provider_id.plugin_id

        with sessionmaker(bind=db.engine).begin() as session:
            lock = f"datasource_provider_create_lock:{tenant_id}_{provider_id}_{CredentialType.API_KEY}"
            with redis_client.lock(lock, timeout=20):
                db_provider_name = name or self.generate_next_datasource_provider_name(
                    session=session,
                    tenant_id=tenant_id,
                    provider_id=provider_id,
                    credential_type=CredentialType.API_KEY,
                )

                # check name is exist
                if (
                    session.scalar(
                        select(func.count(DatasourceProvider.id)).where(
                            DatasourceProvider.tenant_id == tenant_id,
                            DatasourceProvider.plugin_id == plugin_id,
                            DatasourceProvider.provider == provider_name,
                            DatasourceProvider.name == db_provider_name,
                        )
                    )
                    or 0
                ) > 0:
                    raise ValueError("Authorization name is already exists")

                try:
                    current_user = get_current_user()
                    self.provider_manager.validate_provider_credentials(
                        tenant_id=tenant_id,
                        user_id=current_user.id,
                        provider=provider_name,
                        plugin_id=plugin_id,
                        credentials=credentials,
                    )
                except Exception as e:
                    raise ValueError(f"Failed to validate credentials: {str(e)}")

                provider_credential_secret_variables = self.extract_secret_variables(
                    tenant_id=tenant_id, provider_id=f"{provider_id}", credential_type=CredentialType.API_KEY
                )
                for key, value in credentials.items():
                    if key in provider_credential_secret_variables:
                        # if send [__HIDDEN__] in secret input, it will be same as original value
                        credentials[key] = encrypter.encrypt_token(tenant_id, value)
                datasource_provider = DatasourceProvider(
                    tenant_id=tenant_id,
                    name=db_provider_name,
                    provider=provider_name,
                    plugin_id=plugin_id,
                    auth_type=CredentialType.API_KEY,
                    encrypted_credentials=credentials,
                )
                session.add(datasource_provider)

    def extract_secret_variables(self, tenant_id: str, provider_id: str, credential_type: CredentialType) -> list[str]:
        """
        Extract secret input form variables.

        :param credential_form_schemas:
        :return:
        """
        datasource_provider = self.provider_manager.fetch_datasource_provider(
            tenant_id=tenant_id, provider_id=provider_id
        )
        credential_form_schemas = []
        if credential_type == CredentialType.API_KEY:
            credential_form_schemas = list(datasource_provider.declaration.credentials_schema)
        elif credential_type == CredentialType.OAUTH2:
            if not datasource_provider.declaration.oauth_schema:
                raise ValueError("Datasource provider oauth schema not found")
            credential_form_schemas = list(datasource_provider.declaration.oauth_schema.credentials_schema)
        else:
            raise ValueError(f"Invalid credential type: {credential_type}")

        secret_input_form_variables = []
        for credential_form_schema in credential_form_schemas:
            if credential_form_schema.type.value == FormType.SECRET_INPUT:
                secret_input_form_variables.append(credential_form_schema.name)

        return secret_input_form_variables

    def list_datasource_credentials(
        self,
        tenant_id: str,
        provider: str,
        plugin_id: str,
        user: "Account | None" = None,
    ) -> list[dict]:
        """
        list datasource credentials with obfuscated sensitive fields,
        filtered by visibility.

        :param tenant_id: workspace id
        :param provider: provider name
        :param plugin_id: plugin id
        :param user: current user (id + admin flag drive the visibility filter)
        :return:
        """
        from models.credential_permission import CredentialType as CredPermType
        from services.credential_permission_service import CredentialPermissionService

        # Get all provider configurations of the current workspace
        query = select(DatasourceProvider).where(
            DatasourceProvider.tenant_id == tenant_id,
            DatasourceProvider.provider == provider,
            DatasourceProvider.plugin_id == plugin_id,
        )
        if user is not None:
            query = CredentialPermissionService.apply_visibility_filter(
                query,
                model_id_column=DatasourceProvider.id,
                model_user_id_column=DatasourceProvider.user_id,
                model_visibility_column=DatasourceProvider.visibility,
                credential_type=CredPermType.DATASOURCE_PROVIDER,
                user=user,
            )
        datasource_providers: list[DatasourceProvider] = list(db.session.scalars(query).all())
        if not datasource_providers:
            return []
        copy_credentials_list = []
        default_provider = db.session.execute(
            select(DatasourceProvider.id)
            .where(
                DatasourceProvider.tenant_id == tenant_id,
                DatasourceProvider.provider == provider,
                DatasourceProvider.plugin_id == plugin_id,
            )
            .order_by(DatasourceProvider.is_default.desc(), DatasourceProvider.created_at.asc())
        ).first()
        default_provider_id = default_provider.id if default_provider else None
        for datasource_provider in datasource_providers:
            encrypted_credentials = datasource_provider.encrypted_credentials
            # Get provider credential secret variables
            credential_secret_variables = self.extract_secret_variables(
                tenant_id=tenant_id,
                provider_id=f"{plugin_id}/{provider}",
                credential_type=CredentialType.of(datasource_provider.auth_type),
            )

            # Obfuscate provider credentials
            copy_credentials = encrypted_credentials.copy()
            for key, value in copy_credentials.items():
                if key in credential_secret_variables:
                    copy_credentials[key] = encrypter.obfuscated_token(value)
            copy_credentials_list.append(
                {
                    "credential": copy_credentials,
                    "type": datasource_provider.auth_type,
                    "name": datasource_provider.name,
                    "avatar_url": datasource_provider.avatar_url,
                    "id": datasource_provider.id,
                    "is_default": default_provider_id and datasource_provider.id == default_provider_id,
                }
            )

        return copy_credentials_list

    def get_all_datasource_credentials(self, tenant_id: str) -> list[dict]:
        """
        get datasource credentials.

        :return:
        """
        # get all plugin providers
        manager = PluginDatasourceManager()
        datasources = manager.fetch_installed_datasource_providers(tenant_id)
        datasource_credentials = []
        for datasource in datasources:
            datasource_provider_id = DatasourceProviderID(f"{datasource.plugin_id}/{datasource.provider}")
            credentials = self.list_datasource_credentials(
                tenant_id=tenant_id, provider=datasource.provider, plugin_id=datasource.plugin_id
            )
            redirect_uri = (
                f"{dify_config.CONSOLE_API_URL}/console/api/oauth/plugin/{datasource_provider_id}/datasource/callback"
            )
            datasource_credentials.append(
                {
                    "provider": datasource.provider,
                    "plugin_id": datasource.plugin_id,
                    "plugin_unique_identifier": datasource.plugin_unique_identifier,
                    "icon": datasource.declaration.identity.icon,
                    "name": datasource.declaration.identity.name.split("/")[-1],
                    "label": datasource.declaration.identity.label.model_dump(),
                    "description": datasource.declaration.identity.description.model_dump(),
                    "author": datasource.declaration.identity.author,
                    "credentials_list": credentials,
                    "credential_schema": [
                        credential.model_dump() for credential in datasource.declaration.credentials_schema
                    ],
                    "oauth_schema": {
                        "client_schema": [
                            client_schema.model_dump()
                            for client_schema in datasource.declaration.oauth_schema.client_schema
                        ],
                        "credentials_schema": [
                            credential_schema.model_dump()
                            for credential_schema in datasource.declaration.oauth_schema.credentials_schema
                        ],
                        "oauth_custom_client_params": self.get_tenant_oauth_client(
                            tenant_id, datasource_provider_id, mask=True
                        ),
                        "is_oauth_custom_client_enabled": self.is_tenant_oauth_params_enabled(
                            tenant_id, datasource_provider_id
                        ),
                        "is_system_oauth_params_exists": self.is_system_oauth_params_exist(datasource_provider_id),
                        "redirect_uri": redirect_uri,
                    }
                    if datasource.declaration.oauth_schema
                    else None,
                }
            )
        return datasource_credentials

    def get_hard_code_datasource_credentials(self, tenant_id: str) -> list[dict]:
        """
        get hard code datasource credentials.

        :return:
        """
        # get all plugin providers
        manager = PluginDatasourceManager()
        datasources = manager.fetch_installed_datasource_providers(tenant_id)
        datasource_credentials = []
        for datasource in datasources:
            if datasource.plugin_id in [
                "langgenius/firecrawl_datasource",
                "langgenius/notion_datasource",
                "langgenius/jina_datasource",
                "watercrawl/watercrawl_datasource",
            ]:
                datasource_provider_id = DatasourceProviderID(f"{datasource.plugin_id}/{datasource.provider}")
                credentials = self.list_datasource_credentials(
                    tenant_id=tenant_id, provider=datasource.provider, plugin_id=datasource.plugin_id
                )
                redirect_uri = "{}/console/api/oauth/plugin/{}/datasource/callback".format(
                    dify_config.CONSOLE_API_URL, datasource_provider_id
                )
                datasource_credentials.append(
                    {
                        "provider": datasource.provider,
                        "plugin_id": datasource.plugin_id,
                        "plugin_unique_identifier": datasource.plugin_unique_identifier,
                        "icon": datasource.declaration.identity.icon,
                        "name": datasource.declaration.identity.name.split("/")[-1],
                        "label": datasource.declaration.identity.label.model_dump(),
                        "description": datasource.declaration.identity.description.model_dump(),
                        "author": datasource.declaration.identity.author,
                        "credentials_list": credentials,
                        "credential_schema": [
                            credential.model_dump() for credential in datasource.declaration.credentials_schema
                        ],
                        "oauth_schema": {
                            "client_schema": [
                                client_schema.model_dump()
                                for client_schema in datasource.declaration.oauth_schema.client_schema
                            ],
                            "credentials_schema": [
                                credential_schema.model_dump()
                                for credential_schema in datasource.declaration.oauth_schema.credentials_schema
                            ],
                            "oauth_custom_client_params": self.get_tenant_oauth_client(
                                tenant_id, datasource_provider_id, mask=True
                            ),
                            "is_oauth_custom_client_enabled": self.is_tenant_oauth_params_enabled(
                                tenant_id, datasource_provider_id
                            ),
                            "is_system_oauth_params_exists": self.is_system_oauth_params_exist(datasource_provider_id),
                            "redirect_uri": redirect_uri,
                        }
                        if datasource.declaration.oauth_schema
                        else None,
                    }
                )
        return datasource_credentials

    def get_real_datasource_credentials(self, tenant_id: str, provider: str, plugin_id: str) -> list[dict]:
        """
        get datasource credentials.

        :param tenant_id: workspace id
        :param provider_id: provider id
        :return:
        """
        # Get all provider configurations of the current workspace
        datasource_providers: list[DatasourceProvider] = list(
            db.session.scalars(
                select(DatasourceProvider).where(
                    DatasourceProvider.tenant_id == tenant_id,
                    DatasourceProvider.provider == provider,
                    DatasourceProvider.plugin_id == plugin_id,
                )
            ).all()
        )
        if not datasource_providers:
            return []
        copy_credentials_list = []
        for datasource_provider in datasource_providers:
            encrypted_credentials = datasource_provider.encrypted_credentials
            # Get provider credential secret variables
            credential_secret_variables = self.extract_secret_variables(
                tenant_id=tenant_id,
                provider_id=f"{plugin_id}/{provider}",
                credential_type=CredentialType.of(datasource_provider.auth_type),
            )

            # Obfuscate provider credentials
            copy_credentials = encrypted_credentials.copy()
            for key, value in copy_credentials.items():
                if key in credential_secret_variables:
                    copy_credentials[key] = encrypter.decrypt_token(tenant_id, value)
            copy_credentials_list.append(
                {
                    "credentials": copy_credentials,
                    "type": datasource_provider.auth_type,
                }
            )

        return copy_credentials_list

    def update_datasource_credentials(
        self,
        tenant_id: str,
        auth_id: str,
        provider: str,
        plugin_id: str,
        credentials: dict[str, Any] | None,
        name: str | None,
    ) -> None:
        """
        update datasource credentials.
        """

        with sessionmaker(bind=db.engine).begin() as session:
            datasource_provider = session.scalar(
                select(DatasourceProvider)
                .where(
                    DatasourceProvider.tenant_id == tenant_id,
                    DatasourceProvider.id == auth_id,
                    DatasourceProvider.provider == provider,
                    DatasourceProvider.plugin_id == plugin_id,
                )
                .limit(1)
            )
            if not datasource_provider:
                raise ValueError("Datasource provider not found")
            # update name
            if name and name != datasource_provider.name:
                if (
                    session.scalar(
                        select(func.count(DatasourceProvider.id)).where(
                            DatasourceProvider.tenant_id == tenant_id,
                            DatasourceProvider.name == name,
                            DatasourceProvider.provider == provider,
                            DatasourceProvider.plugin_id == plugin_id,
                        )
                    )
                    or 0
                ) > 0:
                    raise ValueError("Authorization name is already exists")
                datasource_provider.name = name

            # update credentials
            if credentials:
                secret_variables = self.extract_secret_variables(
                    tenant_id=tenant_id,
                    provider_id=f"{plugin_id}/{provider}",
                    credential_type=CredentialType.of(datasource_provider.auth_type),
                )
                original_credentials = {
                    key: value if key not in secret_variables else encrypter.decrypt_token(tenant_id, value)
                    for key, value in datasource_provider.encrypted_credentials.items()
                }
                new_credentials = {
                    key: value if value != HIDDEN_VALUE else original_credentials.get(key, UNKNOWN_VALUE)
                    for key, value in credentials.items()
                }
                try:
                    current_user = get_current_user()
                    self.provider_manager.validate_provider_credentials(
                        tenant_id=tenant_id,
                        user_id=current_user.id,
                        provider=provider,
                        plugin_id=plugin_id,
                        credentials=new_credentials,
                    )
                except Exception as e:
                    raise ValueError(f"Failed to validate credentials: {str(e)}")

                encrypted_credentials = {}
                for key, value in new_credentials.items():
                    if key in secret_variables:
                        encrypted_credentials[key] = encrypter.encrypt_token(tenant_id, value)
                    else:
                        encrypted_credentials[key] = value

                datasource_provider.encrypted_credentials = encrypted_credentials

    def remove_datasource_credentials(self, tenant_id: str, auth_id: str, provider: str, plugin_id: str) -> None:
        """
        remove datasource credentials.

        :param tenant_id: workspace id
        :param provider: provider name
        :param plugin_id: plugin id
        :return:
        """
        datasource_provider = db.session.scalar(
            select(DatasourceProvider)
            .where(
                DatasourceProvider.tenant_id == tenant_id,
                DatasourceProvider.id == auth_id,
                DatasourceProvider.provider == provider,
                DatasourceProvider.plugin_id == plugin_id,
            )
            .limit(1)
        )
        if datasource_provider:
            db.session.delete(datasource_provider)
            db.session.commit()
