from __future__ import annotations

import json
import re
import uuid
from collections.abc import Callable, Mapping, Sequence
from datetime import datetime
from decimal import Decimal
from enum import StrEnum, auto
from functools import lru_cache
from typing import TYPE_CHECKING, Any, Literal, NotRequired, TypedDict, cast
from uuid import uuid4

import sqlalchemy as sa
from flask import request
from flask_login import UserMixin  # type: ignore[import-untyped]
from sqlalchemy import BigInteger, Float, Index, PrimaryKeyConstraint, String, exists, func, select, text
from sqlalchemy.orm import Mapped, Session, mapped_column, sessionmaker

from configs import dify_config
from constants import DEFAULT_FILE_NUMBER_LIMITS
from core.tools.signature import sign_tool_file
from extensions.storage.storage_type import StorageType
from graphon.enums import WorkflowExecutionStatus
from graphon.file import FILE_MODEL_IDENTITY, File, FileTransferMethod, FileType
from graphon.file import helpers as file_helpers
from libs.helper import generate_string  # type: ignore[import-not-found]
from libs.url_utils import normalize_api_base_url
from libs.uuid_utils import uuidv7
from models.utils.file_input_compat import build_file_from_input_mapping

from .account import Account, Tenant
from .base import Base, TypeBase, gen_uuidv4_string
from .engine import db
from .enums import (
    ApiTokenType,
    AppMCPServerStatus,
    AppStatus,
    BannerStatus,
    ConversationFromSource,
    ConversationStatus,
    CreatorUserRole,
    CustomizeTokenStrategy,
    FeedbackFromSource,
    FeedbackRating,
    InvokeFrom,
    MessageChainType,
    MessageFileBelongsTo,
    MessageStatus,
    PromptType,
    ProviderQuotaType,
    TagType,
)
from .provider_ids import GenericProviderID
from .types import EnumText, LongText, StringUUID

if TYPE_CHECKING:
    from .workflow import Workflow


# --- TypedDict definitions for structured dict return types ---


@lru_cache(maxsize=1)
def _get_file_access_controller():
    from core.app.file_access import DatabaseFileAccessController

    return DatabaseFileAccessController()


def _resolve_app_tenant_id(app_id: str) -> str:
    resolved_tenant_id = db.session.scalar(select(App.tenant_id).where(App.id == app_id))
    if not resolved_tenant_id:
        raise ValueError(f"Unable to resolve tenant_id for app {app_id}")
    return resolved_tenant_id


def _build_app_tenant_resolver(app_id: str, owner_tenant_id: str | None = None) -> Callable[[], str]:
    resolved_tenant_id = owner_tenant_id

    def resolve_owner_tenant_id() -> str:
        nonlocal resolved_tenant_id
        if resolved_tenant_id is None:
            resolved_tenant_id = _resolve_app_tenant_id(app_id)
        return resolved_tenant_id

    return resolve_owner_tenant_id


class EnabledConfig(TypedDict):
    enabled: bool


class SuggestedQuestionsAfterAnswerModelConfig(TypedDict):
    provider: str
    name: str
    mode: NotRequired[str]
    completion_params: NotRequired[dict[str, Any]]


class SuggestedQuestionsAfterAnswerConfig(TypedDict):
    enabled: bool
    model: NotRequired[SuggestedQuestionsAfterAnswerModelConfig]
    prompt: NotRequired[str]


class EmbeddingModelInfo(TypedDict):
    embedding_provider_name: str
    embedding_model_name: str


class AnnotationReplyDisabledConfig(TypedDict):
    enabled: Literal[False]


class AnnotationReplyEnabledConfig(TypedDict):
    id: str
    enabled: Literal[True]
    score_threshold: float
    embedding_model: EmbeddingModelInfo


AnnotationReplyConfig = AnnotationReplyEnabledConfig | AnnotationReplyDisabledConfig


class SensitiveWordAvoidanceConfig(TypedDict):
    enabled: bool
    type: str
    config: dict[str, Any]


class AgentToolConfig(TypedDict):
    provider_type: str
    provider_id: str
    tool_name: str
    tool_parameters: dict[str, Any]
    plugin_unique_identifier: NotRequired[str | None]
    credential_id: NotRequired[str | None]


class AgentModeConfig(TypedDict):
    enabled: bool
    strategy: str | None
    tools: list[AgentToolConfig | dict[str, Any]]
    prompt: str | None


class ImageUploadConfig(TypedDict):
    enabled: bool
    number_limits: int
    detail: str
    transfer_methods: list[str]


class FileUploadConfig(TypedDict):
    image: ImageUploadConfig


class DeletedToolInfo(TypedDict):
    type: str
    tool_name: str
    provider_id: str


class ExternalDataToolConfig(TypedDict):
    enabled: bool
    variable: str
    type: str
    config: dict[str, Any]


class UserInputFormItemConfig(TypedDict):
    variable: str
    label: str
    description: NotRequired[str]
    required: NotRequired[bool]
    max_length: NotRequired[int]
    options: NotRequired[list[str]]
    default: NotRequired[str]
    type: NotRequired[str]
    config: NotRequired[dict[str, Any]]


# Each item is a single-key dict, e.g. {"text-input": UserInputFormItemConfig}
UserInputFormItem = dict[str, UserInputFormItemConfig]


class DatasetConfigs(TypedDict):
    retrieval_model: str
    datasets: NotRequired[dict[str, Any]]
    top_k: NotRequired[int]
    score_threshold: NotRequired[float]
    score_threshold_enabled: NotRequired[bool]
    reranking_model: NotRequired[dict[str, Any] | None]
    weights: NotRequired[dict[str, Any] | None]
    reranking_enabled: NotRequired[bool]
    reranking_mode: NotRequired[str]
    metadata_filtering_mode: NotRequired[str]
    metadata_model_config: NotRequired[dict[str, Any] | None]
    metadata_filtering_conditions: NotRequired[dict[str, Any] | None]


class ChatPromptMessage(TypedDict):
    text: str
    role: str


class ChatPromptConfig(TypedDict, total=False):
    prompt: list[ChatPromptMessage]


class CompletionPromptText(TypedDict):
    text: str


class ConversationHistoriesRole(TypedDict):
    user_prefix: str
    assistant_prefix: str


class CompletionPromptConfig(TypedDict):
    prompt: CompletionPromptText
    conversation_histories_role: NotRequired[ConversationHistoriesRole]


class ModelConfig(TypedDict):
    provider: str
    name: str
    mode: str
    completion_params: NotRequired[dict[str, Any]]


class AppModelConfigDict(TypedDict):
    opening_statement: str | None
    suggested_questions: list[str]
    suggested_questions_after_answer: SuggestedQuestionsAfterAnswerConfig
    speech_to_text: EnabledConfig
    text_to_speech: EnabledConfig
    retriever_resource: EnabledConfig
    annotation_reply: AnnotationReplyConfig
    more_like_this: EnabledConfig
    sensitive_word_avoidance: SensitiveWordAvoidanceConfig
    external_data_tools: list[ExternalDataToolConfig]
    model: ModelConfig
    user_input_form: list[UserInputFormItem]
    dataset_query_variable: str | None
    pre_prompt: str | None
    agent_mode: AgentModeConfig
    prompt_type: str
    chat_prompt_config: ChatPromptConfig
    completion_prompt_config: CompletionPromptConfig
    dataset_configs: DatasetConfigs
    file_upload: FileUploadConfig
    # Added dynamically in Conversation.model_config
    model_id: NotRequired[str | None]
    provider: NotRequired[str | None]


class ConversationDict(TypedDict):
    id: str
    app_id: str
    app_model_config_id: str | None
    model_provider: str | None
    override_model_configs: str | None
    model_id: str | None
    mode: str
    name: str
    summary: str | None
    inputs: dict[str, Any]
    introduction: str | None
    system_instruction: str | None
    system_instruction_tokens: int
    status: str
    invoke_from: str | None
    from_source: str
    from_end_user_id: str | None
    from_account_id: str | None
    read_at: datetime | None
    read_account_id: str | None
    dialogue_count: int
    created_at: datetime
    updated_at: datetime


class MessageDict(TypedDict):
    id: str
    app_id: str
    conversation_id: str
    model_id: str | None
    inputs: dict[str, Any]
    query: str
    total_price: Decimal | None
    message: dict[str, Any]
    answer: str
    status: str
    error: str | None
    message_metadata: dict[str, Any]
    from_source: str
    from_end_user_id: str | None
    from_account_id: str | None
    created_at: str
    updated_at: str
    agent_based: bool
    workflow_run_id: str | None


class MessageFeedbackDict(TypedDict):
    id: str
    app_id: str
    conversation_id: str
    message_id: str
    rating: str
    content: str | None
    from_source: str
    from_end_user_id: str | None
    from_account_id: str | None
    created_at: str
    updated_at: str


class MessageFileInfo(TypedDict, total=False):
    belongs_to: str | None
    upload_file_id: str | None
    id: str
    tenant_id: str
    type: str
    transfer_method: str
    remote_url: str | None
    related_id: str | None
    filename: str | None
    extension: str | None
    mime_type: str | None
    size: int
    dify_model_identity: str
    url: str | None


class ExtraContentDict(TypedDict, total=False):
    type: str
    workflow_run_id: str


class TraceAppConfigDict(TypedDict):
    id: str
    app_id: str
    tracing_provider: str | None
    tracing_config: dict[str, Any]
    is_active: bool
    created_at: str | None
    updated_at: str | None


class DifySetup(TypeBase):
    __tablename__ = "dify_setups"
    __table_args__ = (sa.PrimaryKeyConstraint("version", name="dify_setup_pkey"),)

    version: Mapped[str] = mapped_column(String(255), nullable=False)
    setup_at: Mapped[datetime] = mapped_column(
        sa.DateTime, nullable=False, server_default=func.current_timestamp(), init=False
    )


class AppMode(StrEnum):
    COMPLETION = "completion"
    WORKFLOW = "workflow"
    CHAT = "chat"
    ADVANCED_CHAT = "advanced-chat"
    AGENT_CHAT = "agent-chat"
    CHANNEL = "channel"
    RAG_PIPELINE = "rag-pipeline"

    @classmethod
    def value_of(cls, value: str) -> AppMode:
        """
        Get value of given mode.

        :param value: mode value
        :return: mode
        """
        for mode in cls:
            if mode.value == value:
                return mode
        raise ValueError(f"invalid mode value {value}")


class IconType(StrEnum):
    IMAGE = auto()
    EMOJI = auto()
    LINK = auto()


class App(Base):
    __tablename__ = "apps"
    __table_args__ = (sa.PrimaryKeyConstraint("id", name="app_pkey"), sa.Index("app_tenant_id_idx", "tenant_id"))

    id: Mapped[str] = mapped_column(StringUUID, default=lambda: str(uuid4()))
    tenant_id: Mapped[str] = mapped_column(StringUUID)
    name: Mapped[str] = mapped_column(String(255))
    description: Mapped[str] = mapped_column(LongText, default=sa.text("''"))
    mode: Mapped[AppMode] = mapped_column(EnumText(AppMode, length=255))
    icon_type: Mapped[IconType | None] = mapped_column(EnumText(IconType, length=255))
    icon = mapped_column(String(255))
    icon_background: Mapped[str | None] = mapped_column(String(255))
    app_model_config_id = mapped_column(StringUUID, nullable=True)
    workflow_id = mapped_column(StringUUID, nullable=True)
    status: Mapped[AppStatus] = mapped_column(
        EnumText(AppStatus, length=255), server_default=sa.text("'normal'"), default=AppStatus.NORMAL
    )
    enable_site: Mapped[bool] = mapped_column(sa.Boolean)
    enable_api: Mapped[bool] = mapped_column(sa.Boolean)
    api_rpm: Mapped[int] = mapped_column(sa.Integer, server_default=sa.text("0"))
    api_rph: Mapped[int] = mapped_column(sa.Integer, server_default=sa.text("0"))
    is_demo: Mapped[bool] = mapped_column(sa.Boolean, server_default=sa.text("false"))
    is_public: Mapped[bool] = mapped_column(sa.Boolean, server_default=sa.text("false"))
    is_universal: Mapped[bool] = mapped_column(sa.Boolean, server_default=sa.text("false"))
    tracing = mapped_column(LongText, nullable=True)
    max_active_requests: Mapped[int | None]
    created_by = mapped_column(StringUUID, nullable=True)
    created_at = mapped_column(sa.DateTime, nullable=False, server_default=func.current_timestamp())
    updated_by = mapped_column(StringUUID, nullable=True)
    updated_at: Mapped[datetime] = mapped_column(
        sa.DateTime, nullable=False, server_default=func.current_timestamp(), onupdate=func.current_timestamp()
    )
    use_icon_as_answer_icon: Mapped[bool] = mapped_column(sa.Boolean, nullable=False, server_default=sa.text("false"))

    @property
    def desc_or_prompt(self) -> str:
        if self.description:
            return self.description
        else:
            app_model_config = self.app_model_config
            if app_model_config:
                pre_prompt = app_model_config.pre_prompt or ""
                # Truncate to 200 characters with ellipsis if using prompt as description
                if len(pre_prompt) > 200:
                    return pre_prompt[:200] + "..."
                return pre_prompt
            else:
                return ""

    @property
    def site(self) -> Site | None:
        return db.session.scalar(select(Site).where(Site.app_id == self.id))

    @property
    def app_model_config(self) -> AppModelConfig | None:
        if self.app_model_config_id:
            return db.session.scalar(select(AppModelConfig).where(AppModelConfig.id == self.app_model_config_id))

        return None

    @property
    def workflow(self) -> Workflow | None:
        if self.workflow_id:
            from .workflow import Workflow

            return db.session.scalar(select(Workflow).where(Workflow.id == self.workflow_id))

        return None

    @property
    def api_base_url(self) -> str:
        base = dify_config.SERVICE_API_URL or request.host_url.rstrip("/")
        return normalize_api_base_url(base)

    @property
    def tenant(self) -> Tenant | None:
        return db.session.scalar(select(Tenant).where(Tenant.id == self.tenant_id))

    @property
    def is_agent(self) -> bool:
        app_model_config = self.app_model_config
        if not app_model_config:
            return False
        if not app_model_config.agent_mode:
            return False

        if app_model_config.agent_mode_dict.get("enabled", False) and app_model_config.agent_mode_dict.get(
            "strategy", ""
        ) in {"function_call", "react"}:
            self.mode = AppMode.AGENT_CHAT
            db.session.commit()
            return True
        return False

    @property
    def mode_compatible_with_agent(self) -> str:
        if self.mode == AppMode.CHAT and self.is_agent:
            return AppMode.AGENT_CHAT

        return str(self.mode)

    @property
    def deleted_tools(self) -> list[DeletedToolInfo]:
        from core.plugin.plugin_service import PluginService
        from core.tools.tool_manager import ToolManager, ToolProviderType

        # get agent mode tools
        app_model_config = self.app_model_config
        if not app_model_config:
            return []

        if not app_model_config.agent_mode:
            return []

        agent_mode = app_model_config.agent_mode_dict
        tools = agent_mode.get("tools", [])

        api_provider_ids: list[str] = []

        builtin_provider_ids: list[GenericProviderID] = []

        for tool in tools:
            keys = list(tool.keys())
            if len(keys) >= 4:
                provider_type = tool.get("provider_type", "")
                provider_id = tool.get("provider_id", "")
                if provider_type == ToolProviderType.API:
                    try:
                        uuid.UUID(provider_id)
                    except Exception:
                        continue
                    api_provider_ids.append(provider_id)
                if provider_type == ToolProviderType.BUILT_IN:
                    try:
                        # check if it's hardcoded
                        try:
                            ToolManager.get_hardcoded_provider(provider_id)
                            is_hardcoded = True
                        except Exception:
                            is_hardcoded = False

                        provider_id = GenericProviderID(provider_id, is_hardcoded)
                    except Exception:
                        continue

                    builtin_provider_ids.append(provider_id)

        if not api_provider_ids and not builtin_provider_ids:
            return []

        with sessionmaker(db.engine).begin() as session:
            if api_provider_ids:
                existing_api_providers = [
                    str(api_provider.id)
                    for api_provider in session.execute(
                        text("SELECT id FROM tool_api_providers WHERE id IN :provider_ids"),
                        {"provider_ids": tuple(api_provider_ids)},
                    ).fetchall()
                ]
            else:
                existing_api_providers = []

        if builtin_provider_ids:
            # get the non-hardcoded builtin providers
            non_hardcoded_builtin_providers = [
                provider_id for provider_id in builtin_provider_ids if not provider_id.is_hardcoded
            ]
            if non_hardcoded_builtin_providers:
                existence = list(PluginService.check_tools_existence(self.tenant_id, non_hardcoded_builtin_providers))
            else:
                existence = []
            # add the hardcoded builtin providers
            existence.extend([True] * (len(builtin_provider_ids) - len(non_hardcoded_builtin_providers)))
            builtin_provider_ids = non_hardcoded_builtin_providers + [
                provider_id for provider_id in builtin_provider_ids if provider_id.is_hardcoded
            ]
        else:
            existence = []

        existing_builtin_providers = {
            provider_id.provider_name: existence[i] for i, provider_id in enumerate(builtin_provider_ids)
        }

        deleted_tools: list[DeletedToolInfo] = []

        for tool in tools:
            keys = list(tool.keys())
            if len(keys) >= 4:
                provider_type = tool.get("provider_type", "")
                provider_id = tool.get("provider_id", "")

                if provider_type == ToolProviderType.API:
                    if provider_id not in existing_api_providers:
                        deleted_tools.append(
                            {
                                "type": ToolProviderType.API,
                                "tool_name": tool["tool_name"],
                                "provider_id": provider_id,
                            }
                        )

                if provider_type == ToolProviderType.BUILT_IN:
                    generic_provider_id = GenericProviderID(provider_id)

                    if not existing_builtin_providers[generic_provider_id.provider_name]:
                        deleted_tools.append(
                            {
                                "type": ToolProviderType.BUILT_IN,
                                "tool_name": tool["tool_name"],
                                "provider_id": provider_id,  # use the original one
                            }
                        )

        return deleted_tools

    @property
    def tags(self) -> Sequence[Tag]:
        tags = db.session.scalars(
            select(Tag)
            .join(TagBinding, Tag.id == TagBinding.tag_id)
            .where(
                TagBinding.target_id == self.id,
                TagBinding.tenant_id == self.tenant_id,
                Tag.tenant_id == self.tenant_id,
                Tag.type == "app",
            )
        ).all()

        return tags or []

    @property
    def author_name(self) -> str | None:
        if self.created_by:
            account = db.session.scalar(select(Account).where(Account.id == self.created_by))
            if account:
                return account.name

        return None


class AppModelConfig(TypeBase):
    __tablename__ = "app_model_configs"
    __table_args__ = (sa.PrimaryKeyConstraint("id", name="app_model_config_pkey"), sa.Index("app_app_id_idx", "app_id"))

    id: Mapped[str] = mapped_column(
        StringUUID, insert_default=lambda: str(uuid4()), default_factory=lambda: str(uuid4()), init=False
    )
    app_id: Mapped[str] = mapped_column(StringUUID, nullable=False)
    provider: Mapped[str | None] = mapped_column(String(255), nullable=True, default=None)
    model_id: Mapped[str | None] = mapped_column(String(255), nullable=True, default=None)
    configs: Mapped[Any | None] = mapped_column(sa.JSON, nullable=True, default=None)
    created_by: Mapped[str | None] = mapped_column(StringUUID, nullable=True, default=None)
    created_at: Mapped[datetime] = mapped_column(
        sa.DateTime, nullable=False, server_default=func.current_timestamp(), init=False
    )
    updated_by: Mapped[str | None] = mapped_column(StringUUID, nullable=True, default=None)
    updated_at: Mapped[datetime] = mapped_column(
        sa.DateTime,
        nullable=False,
        server_default=func.current_timestamp(),
        onupdate=func.current_timestamp(),
        init=False,
    )
    opening_statement: Mapped[str | None] = mapped_column(LongText, default=None)
    suggested_questions: Mapped[str | None] = mapped_column(LongText, default=None)
    suggested_questions_after_answer: Mapped[str | None] = mapped_column(LongText, default=None)
    speech_to_text: Mapped[str | None] = mapped_column(LongText, default=None)
    text_to_speech: Mapped[str | None] = mapped_column(LongText, default=None)
    more_like_this: Mapped[str | None] = mapped_column(LongText, default=None)
    model: Mapped[str | None] = mapped_column(LongText, default=None)
    user_input_form: Mapped[str | None] = mapped_column(LongText, default=None)
    dataset_query_variable: Mapped[str | None] = mapped_column(String(255), default=None)
    pre_prompt: Mapped[str | None] = mapped_column(LongText, default=None)
    agent_mode: Mapped[str | None] = mapped_column(LongText, default=None)
    sensitive_word_avoidance: Mapped[str | None] = mapped_column(LongText, default=None)
    retriever_resource: Mapped[str | None] = mapped_column(LongText, default=None)
    prompt_type: Mapped[PromptType] = mapped_column(
        EnumText(PromptType, length=255),
        nullable=False,
        server_default=sa.text("'simple'"),
        default=PromptType.SIMPLE,
    )
    chat_prompt_config: Mapped[str | None] = mapped_column(LongText, default=None)
    completion_prompt_config: Mapped[str | None] = mapped_column(LongText, default=None)
    dataset_configs: Mapped[str | None] = mapped_column(LongText, default=None)
    external_data_tools: Mapped[str | None] = mapped_column(LongText, default=None)
    file_upload: Mapped[str | None] = mapped_column(LongText, default=None)

    @property
    def app(self) -> App | None:
        return db.session.scalar(select(App).where(App.id == self.app_id))

    @property
    def model_dict(self) -> ModelConfig:
        return cast(ModelConfig, json.loads(self.model) if self.model else {})

    @property
    def suggested_questions_list(self) -> list[str]:
        return json.loads(self.suggested_questions) if self.suggested_questions else []

    def _get_enabled_config(self, value: str | None, *, default_enabled: bool = False) -> EnabledConfig:
        return cast(EnabledConfig, json.loads(value) if value else {"enabled": default_enabled})

    @property
    def suggested_questions_after_answer_dict(self) -> SuggestedQuestionsAfterAnswerConfig:
        return cast(
            SuggestedQuestionsAfterAnswerConfig,
            json.loads(self.suggested_questions_after_answer)
            if self.suggested_questions_after_answer
            else {"enabled": False},
        )

    @property
    def speech_to_text_dict(self) -> EnabledConfig:
        return self._get_enabled_config(self.speech_to_text)

    @property
    def text_to_speech_dict(self) -> EnabledConfig:
        return self._get_enabled_config(self.text_to_speech)

    @property
    def retriever_resource_dict(self) -> EnabledConfig:
        return self._get_enabled_config(self.retriever_resource, default_enabled=True)

    @property
    def annotation_reply_dict(self) -> AnnotationReplyConfig:
        annotation_setting = db.session.scalar(
            select(AppAnnotationSetting).where(AppAnnotationSetting.app_id == self.app_id)
        )
        if annotation_setting:
            collection_binding_detail = annotation_setting.collection_binding_detail
            if not collection_binding_detail:
                raise ValueError("Collection binding detail not found")

            return {
                "id": annotation_setting.id,
                "enabled": True,
                "score_threshold": annotation_setting.score_threshold,
                "embedding_model": {
                    "embedding_provider_name": collection_binding_detail.provider_name,
                    "embedding_model_name": collection_binding_detail.model_name,
                },
            }

        else:
            return {"enabled": False}

    @property
    def more_like_this_dict(self) -> EnabledConfig:
        return self._get_enabled_config(self.more_like_this)

    @property
    def sensitive_word_avoidance_dict(self) -> SensitiveWordAvoidanceConfig:
        return cast(
            SensitiveWordAvoidanceConfig,
            json.loads(self.sensitive_word_avoidance)
            if self.sensitive_word_avoidance
            else {"enabled": False, "type": "", "config": {}},
        )

    @property
    def external_data_tools_list(self) -> list[ExternalDataToolConfig]:
        return json.loads(self.external_data_tools) if self.external_data_tools else []

    @property
    def user_input_form_list(self) -> list[UserInputFormItem]:
        return json.loads(self.user_input_form) if self.user_input_form else []

    @property
    def agent_mode_dict(self) -> AgentModeConfig:
        return cast(
            AgentModeConfig,
            json.loads(self.agent_mode)
            if self.agent_mode
            else {"enabled": False, "strategy": None, "tools": [], "prompt": None},
        )

    @property
    def chat_prompt_config_dict(self) -> ChatPromptConfig:
        return cast(ChatPromptConfig, json.loads(self.chat_prompt_config) if self.chat_prompt_config else {})

    @property
    def completion_prompt_config_dict(self) -> CompletionPromptConfig:
        return cast(
            CompletionPromptConfig,
            json.loads(self.completion_prompt_config) if self.completion_prompt_config else {},
        )

    @property
    def dataset_configs_dict(self) -> DatasetConfigs:
        if self.dataset_configs:
            dataset_configs = json.loads(self.dataset_configs)
            if "retrieval_model" not in dataset_configs:
                return {"retrieval_model": "single"}
            else:
                return cast(DatasetConfigs, dataset_configs)
        return {
            "retrieval_model": "multiple",
        }

    @property
    def file_upload_dict(self) -> FileUploadConfig:
        return cast(
            FileUploadConfig,
            json.loads(self.file_upload)
            if self.file_upload
            else {
                "image": {
                    "enabled": False,
                    "number_limits": DEFAULT_FILE_NUMBER_LIMITS,
                    "detail": "high",
                    "transfer_methods": ["remote_url", "local_file"],
                }
            },
        )

    def to_dict(self) -> AppModelConfigDict:
        return {
            "opening_statement": self.opening_statement,
            "suggested_questions": self.suggested_questions_list,
            "suggested_questions_after_answer": self.suggested_questions_after_answer_dict,
            "speech_to_text": self.speech_to_text_dict,
            "text_to_speech": self.text_to_speech_dict,
            "retriever_resource": self.retriever_resource_dict,
            "annotation_reply": self.annotation_reply_dict,
            "more_like_this": self.more_like_this_dict,
            "sensitive_word_avoidance": self.sensitive_word_avoidance_dict,
            "external_data_tools": self.external_data_tools_list,
            "model": self.model_dict,
            "user_input_form": self.user_input_form_list,
            "dataset_query_variable": self.dataset_query_variable,
            "pre_prompt": self.pre_prompt,
            "agent_mode": self.agent_mode_dict,
            "prompt_type": self.prompt_type.value if isinstance(self.prompt_type, PromptType) else self.prompt_type,
            "chat_prompt_config": self.chat_prompt_config_dict,
            "completion_prompt_config": self.completion_prompt_config_dict,
            "dataset_configs": self.dataset_configs_dict,
            "file_upload": self.file_upload_dict,
        }

    @staticmethod
    def _dump_optional(value: Any) -> str | None:
        return json.dumps(value) if value else None

    def from_model_config_dict(self, model_config: AppModelConfigDict):
        self.opening_statement = model_config.get("opening_statement")
        self.suggested_questions = self._dump_optional(model_config.get("suggested_questions"))
        self.suggested_questions_after_answer = self._dump_optional(
            model_config.get("suggested_questions_after_answer")
        )
        self.speech_to_text = self._dump_optional(model_config.get("speech_to_text"))
        self.text_to_speech = self._dump_optional(model_config.get("text_to_speech"))
        self.more_like_this = self._dump_optional(model_config.get("more_like_this"))
        self.sensitive_word_avoidance = self._dump_optional(model_config.get("sensitive_word_avoidance"))
        self.external_data_tools = self._dump_optional(model_config.get("external_data_tools"))
        self.model = self._dump_optional(model_config.get("model"))
        self.user_input_form = self._dump_optional(model_config.get("user_input_form"))
        self.dataset_query_variable = model_config.get("dataset_query_variable")
        self.pre_prompt = model_config.get("pre_prompt")
        self.agent_mode = self._dump_optional(model_config.get("agent_mode"))
        self.retriever_resource = self._dump_optional(model_config.get("retriever_resource"))
        self.prompt_type = PromptType(model_config.get("prompt_type", "simple"))
        self.chat_prompt_config = self._dump_optional(model_config.get("chat_prompt_config"))
        self.completion_prompt_config = self._dump_optional(model_config.get("completion_prompt_config"))
        self.dataset_configs = self._dump_optional(model_config.get("dataset_configs"))
        self.file_upload = self._dump_optional(model_config.get("file_upload"))
        return self


class RecommendedApp(TypeBase):
    __tablename__ = "recommended_apps"
    __table_args__ = (
        sa.PrimaryKeyConstraint("id", name="recommended_app_pkey"),
        sa.Index("recommended_app_app_id_idx", "app_id"),
        sa.Index("recommended_app_is_listed_idx", "is_listed", "language"),
    )

    id: Mapped[str] = mapped_column(
        StringUUID,
        primary_key=True,
        insert_default=lambda: str(uuid4()),
        default_factory=lambda: str(uuid4()),
        init=False,
    )
    app_id: Mapped[str] = mapped_column(StringUUID, nullable=False)
    description: Mapped[Any] = mapped_column(sa.JSON, nullable=False)
    copyright: Mapped[str] = mapped_column(String(255), nullable=False)
    privacy_policy: Mapped[str] = mapped_column(String(255), nullable=False)
    category: Mapped[str] = mapped_column(String(255), nullable=False)
    categories: Mapped[list[str] | None] = mapped_column(sa.JSON, nullable=True, default=None)
    custom_disclaimer: Mapped[str] = mapped_column(LongText, default="")
    position: Mapped[int] = mapped_column(sa.Integer, nullable=False, default=0)
    is_listed: Mapped[bool] = mapped_column(sa.Boolean, nullable=False, default=True)
    install_count: Mapped[int] = mapped_column(sa.Integer, nullable=False, default=0)
    language: Mapped[str] = mapped_column(
        String(255),
        nullable=False,
        server_default=sa.text("'en-US'"),
        default="en-US",
    )
    created_at: Mapped[datetime] = mapped_column(
        sa.DateTime, nullable=False, server_default=func.current_timestamp(), init=False
    )
    updated_at: Mapped[datetime] = mapped_column(
        sa.DateTime,
        nullable=False,
        server_default=func.current_timestamp(),
        onupdate=func.current_timestamp(),
        init=False,
    )

    @property
    def app(self) -> App | None:
        return db.session.scalar(select(App).where(App.id == self.app_id))


class InstalledApp(TypeBase):
    __tablename__ = "installed_apps"
    __table_args__ = (
        sa.PrimaryKeyConstraint("id", name="installed_app_pkey"),
        sa.Index("installed_app_tenant_id_idx", "tenant_id"),
        sa.Index("installed_app_app_id_idx", "app_id"),
        sa.UniqueConstraint("tenant_id", "app_id", name="unique_tenant_app"),
    )

    id: Mapped[str] = mapped_column(
        StringUUID, insert_default=lambda: str(uuid4()), default_factory=lambda: str(uuid4()), init=False
    )
    tenant_id: Mapped[str] = mapped_column(StringUUID, nullable=False)
    app_id: Mapped[str] = mapped_column(StringUUID, nullable=False)
    app_owner_tenant_id: Mapped[str] = mapped_column(StringUUID, nullable=False)
    position: Mapped[int] = mapped_column(sa.Integer, nullable=False, default=0)
    is_pinned: Mapped[bool] = mapped_column(sa.Boolean, nullable=False, server_default=sa.text("false"), default=False)
    last_used_at: Mapped[datetime | None] = mapped_column(sa.DateTime, nullable=True, default=None)
    created_at: Mapped[datetime] = mapped_column(
        sa.DateTime, nullable=False, server_default=func.current_timestamp(), init=False
    )

    @property
    def app(self) -> App | None:
        return db.session.scalar(select(App).where(App.id == self.app_id))

    @property
    def tenant(self) -> Tenant | None:
        return db.session.scalar(select(Tenant).where(Tenant.id == self.tenant_id))


class TrialApp(TypeBase):
    __tablename__ = "trial_apps"
    __table_args__ = (
        sa.PrimaryKeyConstraint("id", name="trial_app_pkey"),
        sa.Index("trial_app_app_id_idx", "app_id"),
        sa.Index("trial_app_tenant_id_idx", "tenant_id"),
        sa.UniqueConstraint("app_id", name="unique_trail_app_id"),
    )

    id: Mapped[str] = mapped_column(
        StringUUID, insert_default=gen_uuidv4_string, default_factory=gen_uuidv4_string, init=False
    )
    app_id: Mapped[str] = mapped_column(StringUUID, nullable=False)
    tenant_id: Mapped[str] = mapped_column(StringUUID, nullable=False)
    created_at: Mapped[datetime] = mapped_column(
        sa.DateTime, nullable=False, server_default=func.current_timestamp(), init=False
    )
    trial_limit: Mapped[int] = mapped_column(sa.Integer, nullable=False, default=3)

    @property
    def app(self) -> App | None:
        return db.session.scalar(select(App).where(App.id == self.app_id))


class AccountTrialAppRecord(TypeBase):
    __tablename__ = "account_trial_app_records"
    __table_args__ = (
        sa.PrimaryKeyConstraint("id", name="user_trial_app_pkey"),
        sa.Index("account_trial_app_record_account_id_idx", "account_id"),
        sa.Index("account_trial_app_record_app_id_idx", "app_id"),
        sa.UniqueConstraint("account_id", "app_id", name="unique_account_trial_app_record"),
    )
    id: Mapped[str] = mapped_column(
        StringUUID, insert_default=gen_uuidv4_string, default_factory=gen_uuidv4_string, init=False
    )
    account_id: Mapped[str] = mapped_column(StringUUID, nullable=False)
    app_id: Mapped[str] = mapped_column(StringUUID, nullable=False)
    count: Mapped[int] = mapped_column(sa.Integer, nullable=False, default=0)
    created_at: Mapped[datetime] = mapped_column(
        sa.DateTime, nullable=False, server_default=func.current_timestamp(), init=False
    )

    @property
    def app(self) -> App | None:
        return db.session.scalar(select(App).where(App.id == self.app_id))

    @property
    def user(self) -> Account | None:
        return db.session.scalar(select(Account).where(Account.id == self.account_id))


class ExporleBanner(TypeBase):
    __tablename__ = "exporle_banners"
    __table_args__ = (sa.PrimaryKeyConstraint("id", name="exporler_banner_pkey"),)
    id: Mapped[str] = mapped_column(
        StringUUID, insert_default=gen_uuidv4_string, default_factory=gen_uuidv4_string, init=False
    )
    content: Mapped[dict[str, Any]] = mapped_column(sa.JSON, nullable=False)
    link: Mapped[str] = mapped_column(String(255), nullable=False)
    sort: Mapped[int] = mapped_column(sa.Integer, nullable=False)
    status: Mapped[BannerStatus] = mapped_column(
        EnumText(BannerStatus, length=255),
        nullable=False,
        server_default=sa.text("'enabled'::character varying"),
        default=BannerStatus.ENABLED,
    )
    created_at: Mapped[datetime] = mapped_column(
        sa.DateTime, nullable=False, server_default=func.current_timestamp(), init=False
    )
    language: Mapped[str] = mapped_column(
        String(255), nullable=False, server_default=sa.text("'en-US'::character varying"), default="en-US"
    )


class OAuthProviderApp(TypeBase):
    """
    Globally shared OAuth provider app information.
    Only for Dify Cloud.
    """

    __tablename__ = "oauth_provider_apps"
    __table_args__ = (
        sa.PrimaryKeyConstraint("id", name="oauth_provider_app_pkey"),
        sa.Index("oauth_provider_app_client_id_idx", "client_id"),
    )

    id: Mapped[str] = mapped_column(
        StringUUID, insert_default=lambda: str(uuidv7()), default_factory=lambda: str(uuidv7()), init=False
    )
    app_icon: Mapped[str] = mapped_column(String(255), nullable=False)
    client_id: Mapped[str] = mapped_column(String(255), nullable=False)
    client_secret: Mapped[str] = mapped_column(String(255), nullable=False)
    app_label: Mapped[dict[str, Any]] = mapped_column(sa.JSON, nullable=False, default_factory=dict)
    redirect_uris: Mapped[list] = mapped_column(sa.JSON, nullable=False, default_factory=list)
    scope: Mapped[str] = mapped_column(
        String(255),
        nullable=False,
        server_default=sa.text("'read:name read:email read:avatar read:interface_language read:timezone'"),
        default="read:name read:email read:avatar read:interface_language read:timezone",
    )
    created_at: Mapped[datetime] = mapped_column(
        sa.DateTime, nullable=False, server_default=func.current_timestamp(), init=False
    )


class Conversation(Base):
    __tablename__ = "conversations"
    __table_args__ = (
        sa.PrimaryKeyConstraint("id", name="conversation_pkey"),
        sa.Index("conversation_app_from_user_idx", "app_id", "from_source", "from_end_user_id"),
        sa.Index(
            "conversation_app_created_at_idx",
            "app_id",
            sa.text("created_at DESC"),
            postgresql_where=sa.text("is_deleted IS false"),
        ),
        sa.Index(
            "conversation_app_updated_at_idx",
            "app_id",
            sa.text("updated_at DESC"),
            postgresql_where=sa.text("is_deleted IS false"),
        ),
    )

    id: Mapped[str] = mapped_column(StringUUID, default=lambda: str(uuid4()))
    app_id = mapped_column(StringUUID, nullable=False)
    app_model_config_id = mapped_column(StringUUID, nullable=True)
    model_provider = mapped_column(String(255), nullable=True)
    override_model_configs = mapped_column(LongText)
    model_id = mapped_column(String(255), nullable=True)
    mode: Mapped[AppMode] = mapped_column(EnumText(AppMode, length=255))
    name: Mapped[str] = mapped_column(String(255), nullable=False)
    summary = mapped_column(LongText)
    _inputs: Mapped[dict[str, Any]] = mapped_column("inputs", sa.JSON)
    introduction = mapped_column(LongText)
    system_instruction = mapped_column(LongText)
    system_instruction_tokens: Mapped[int] = mapped_column(sa.Integer, nullable=False, server_default=sa.text("0"))
    status: Mapped[ConversationStatus] = mapped_column(
        EnumText(ConversationStatus, length=255), nullable=False, default=ConversationStatus.NORMAL
    )

    # The `invoke_from` records how the conversation is created.
    #
    # Its value corresponds to the members of `InvokeFrom`.
    # (api/core/app/entities/app_invoke_entities.py)
    invoke_from: Mapped[InvokeFrom | None] = mapped_column(EnumText(InvokeFrom, length=255), nullable=True)

    # ref: ConversationSource.
    from_source: Mapped[ConversationFromSource] = mapped_column(
        EnumText(ConversationFromSource, length=255), nullable=False
    )
    from_end_user_id = mapped_column(StringUUID)
    from_account_id = mapped_column(StringUUID)
    read_at = mapped_column(sa.DateTime)
    read_account_id = mapped_column(StringUUID)
    dialogue_count: Mapped[int] = mapped_column(default=0)
    created_at = mapped_column(sa.DateTime, nullable=False, server_default=func.current_timestamp())
    updated_at = mapped_column(
        sa.DateTime, nullable=False, server_default=func.current_timestamp(), onupdate=func.current_timestamp()
    )

    messages = db.relationship("Message", backref="conversation", lazy="select", passive_deletes="all")
    message_annotations = db.relationship(
        lambda: MessageAnnotation, backref="conversation", lazy="select", passive_deletes="all"
    )

    is_deleted: Mapped[bool] = mapped_column(sa.Boolean, nullable=False, server_default=sa.text("false"))

    @property
    def inputs(self) -> dict[str, Any]:
        inputs = self._inputs.copy()
        # Compatibility bridge: stored input payloads may come from before or after the
        # graph-layer file refactor. Newer rows may omit `tenant_id`, so keep tenant
        # resolution at the SQLAlchemy model boundary instead of pushing ownership back
        # into `graphon.file.File`.
        tenant_resolver = _build_app_tenant_resolver(
            app_id=self.app_id,
            owner_tenant_id=cast(str | None, getattr(self, "_owner_tenant_id", None)),
        )

        # Convert file mapping to File object
        for key, value in inputs.items():
            if (
                isinstance(value, dict)
                and cast(dict[str, Any], value).get("dify_model_identity") == FILE_MODEL_IDENTITY
            ):
                value_dict = cast(dict[str, Any], value)
                inputs[key] = build_file_from_input_mapping(
                    file_mapping=value_dict,
                    tenant_resolver=tenant_resolver,
                )
            elif isinstance(value, list):
                value_list = cast(list[Any], value)
                if all(
                    isinstance(item, dict)
                    and cast(dict[str, Any], item).get("dify_model_identity") == FILE_MODEL_IDENTITY
                    for item in value_list
                ):
                    file_list: list[File] = []
                    for item in value_list:
                        if not isinstance(item, dict):
                            continue
                        item_dict = cast(dict[str, Any], item)
                        file_list.append(
                            build_file_from_input_mapping(
                                file_mapping=item_dict,
                                tenant_resolver=tenant_resolver,
                            )
                        )
                    inputs[key] = file_list

        return inputs

    @inputs.setter
    def inputs(self, value: Mapping[str, Any]):
        inputs = dict(value)
        for k, v in inputs.items():
            if isinstance(v, File):
                inputs[k] = v.model_dump()
            elif isinstance(v, list):
                v_list = cast(list[Any], v)
                if all(isinstance(item, File) for item in v_list):
                    inputs[k] = [item.model_dump() for item in v_list if isinstance(item, File)]
        self._inputs = inputs

    @property
    def model_config(self) -> AppModelConfigDict:
        model_config = cast(AppModelConfigDict, {})
        app_model_config: AppModelConfig | None = None

        if self.mode == AppMode.ADVANCED_CHAT:
            if self.override_model_configs:
                override_model_configs = json.loads(self.override_model_configs)
                model_config = cast(AppModelConfigDict, override_model_configs)
        else:
            if self.override_model_configs:
                override_model_configs = json.loads(self.override_model_configs)

                if "model" in override_model_configs:
                    # where is app_id?
                    app_model_config = AppModelConfig(app_id=self.app_id).from_model_config_dict(
                        cast(AppModelConfigDict, override_model_configs)
                    )
                    model_config = app_model_config.to_dict()
                else:
                    model_config["configs"] = override_model_configs  # type: ignore[typeddict-unknown-key]
            else:
                app_model_config = db.session.scalar(
                    select(AppModelConfig).where(AppModelConfig.id == self.app_model_config_id)
                )
                if app_model_config:
                    model_config = app_model_config.to_dict()

        model_config["model_id"] = self.model_id
        model_config["provider"] = self.model_provider

        return model_config

    @property
    def summary_or_query(self):
        if self.summary:
            return self.summary
        else:
            first_message = self.first_message
            if first_message:
                return first_message.query
            else:
                return ""

    @property
    def annotated(self):
        return (
            db.session.scalar(
                select(func.count(MessageAnnotation.id)).where(MessageAnnotation.conversation_id == self.id)
            )
            or 0
        ) > 0

    @property
    def annotation(self):
        return db.session.scalar(select(MessageAnnotation).where(MessageAnnotation.conversation_id == self.id).limit(1))

    @property
    def message_count(self):
        return db.session.scalar(select(func.count(Message.id)).where(Message.conversation_id == self.id)) or 0

    @property
    def user_feedback_stats(self):
        like = (
            db.session.scalar(
                select(func.count(MessageFeedback.id)).where(
                    MessageFeedback.conversation_id == self.id,
                    MessageFeedback.from_source == "user",
                    MessageFeedback.rating == FeedbackRating.LIKE,
                )
            )
            or 0
        )

        dislike = (
            db.session.scalar(
                select(func.count(MessageFeedback.id)).where(
                    MessageFeedback.conversation_id == self.id,
                    MessageFeedback.from_source == "user",
                    MessageFeedback.rating == FeedbackRating.DISLIKE,
                )
            )
            or 0
        )

        return {"like": like, "dislike": dislike}

    @property
    def admin_feedback_stats(self):
        like = (
            db.session.scalar(
                select(func.count(MessageFeedback.id)).where(
                    MessageFeedback.conversation_id == self.id,
                    MessageFeedback.from_source == "admin",
                    MessageFeedback.rating == FeedbackRating.LIKE,
                )
            )
            or 0
        )

        dislike = (
            db.session.scalar(
                select(func.count(MessageFeedback.id)).where(
                    MessageFeedback.conversation_id == self.id,
                    MessageFeedback.from_source == "admin",
                    MessageFeedback.rating == FeedbackRating.DISLIKE,
                )
            )
            or 0
        )

        return {"like": like, "dislike": dislike}

    @property
    def status_count(self):
        from models.workflow import WorkflowRun

        # Get all messages with workflow_run_id for this conversation
        messages = db.session.scalars(
            select(Message).where(Message.conversation_id == self.id, Message.workflow_run_id.isnot(None))
        ).all()

        if not messages:
            return None

        # Batch load all workflow runs in a single query, filtered by this conversation's app_id
        workflow_run_ids = [msg.workflow_run_id for msg in messages if msg.workflow_run_id]
        workflow_runs = {}

        if workflow_run_ids:
            workflow_runs_query = db.session.scalars(
                select(WorkflowRun).where(
                    WorkflowRun.id.in_(workflow_run_ids),
                    WorkflowRun.app_id == self.app_id,  # Filter by this conversation's app_id
                )
            ).all()
            workflow_runs = {run.id: run for run in workflow_runs_query}

        status_counts = {
            WorkflowExecutionStatus.RUNNING: 0,
            WorkflowExecutionStatus.SUCCEEDED: 0,
            WorkflowExecutionStatus.FAILED: 0,
            WorkflowExecutionStatus.STOPPED: 0,
            WorkflowExecutionStatus.PARTIAL_SUCCEEDED: 0,
            WorkflowExecutionStatus.PAUSED: 0,
        }

        for message in messages:
            # Guard against None to satisfy type checker and avoid invalid dict lookups
            if message.workflow_run_id is None:
                continue
            workflow_run = workflow_runs.get(message.workflow_run_id)
            if not workflow_run:
                continue

            try:
                status_counts[WorkflowExecutionStatus(workflow_run.status)] += 1
            except (ValueError, KeyError):
                # Handle invalid status values gracefully
                pass

        return {
            "success": status_counts[WorkflowExecutionStatus.SUCCEEDED],
            "failed": status_counts[WorkflowExecutionStatus.FAILED],
            "partial_success": status_counts[WorkflowExecutionStatus.PARTIAL_SUCCEEDED],
            "paused": status_counts[WorkflowExecutionStatus.PAUSED],
        }

    @property
    def first_message(self):
        return db.session.scalar(
            select(Message).where(Message.conversation_id == self.id).order_by(Message.created_at.asc())
        )

    @property
    def app(self) -> App | None:
        with Session(db.engine, expire_on_commit=False) as session:
            return session.scalar(select(App).where(App.id == self.app_id))

    @property
    def from_end_user_session_id(self):
        if self.from_end_user_id:
            end_user = db.session.scalar(select(EndUser).where(EndUser.id == self.from_end_user_id))
            if end_user:
                return end_user.session_id

        return None

    @property
    def from_account_name(self) -> str | None:
        if self.from_account_id:
            account = db.session.scalar(select(Account).where(Account.id == self.from_account_id))
            if account:
                return account.name

        return None

    @property
    def in_debug_mode(self) -> bool:
        return self.override_model_configs is not None

    def to_dict(self) -> ConversationDict:
        return {
            "id": self.id,
            "app_id": self.app_id,
            "app_model_config_id": self.app_model_config_id,
            "model_provider": self.model_provider,
            "override_model_configs": self.override_model_configs,
            "model_id": self.model_id,
            "mode": self.mode,
            "name": self.name,
            "summary": self.summary,
            "inputs": self.inputs,
            "introduction": self.introduction,
            "system_instruction": self.system_instruction,
            "system_instruction_tokens": self.system_instruction_tokens,
            "status": self.status,
            "invoke_from": self.invoke_from,
            "from_source": self.from_source,
            "from_end_user_id": self.from_end_user_id,
            "from_account_id": self.from_account_id,
            "read_at": self.read_at,
            "read_account_id": self.read_account_id,
            "dialogue_count": self.dialogue_count,
            "created_at": self.created_at,
            "updated_at": self.updated_at,
        }


class Message(Base):
    __tablename__ = "messages"
    __table_args__ = (
        PrimaryKeyConstraint("id", name="message_pkey"),
        Index("message_app_id_idx", "app_id", "created_at"),
        Index("message_conversation_id_idx", "conversation_id"),
        Index("message_end_user_idx", "app_id", "from_source", "from_end_user_id"),
        Index("message_account_idx", "app_id", "from_source", "from_account_id"),
        Index("message_workflow_run_id_idx", "conversation_id", "workflow_run_id"),
        Index("message_app_mode_idx", "app_mode"),
        Index("message_created_at_id_idx", "created_at", "id"),
    )

    id: Mapped[str] = mapped_column(StringUUID, default=lambda: str(uuid4()))
    app_id: Mapped[str] = mapped_column(StringUUID, nullable=False)
    model_provider: Mapped[str | None] = mapped_column(String(255), nullable=True)
    model_id: Mapped[str | None] = mapped_column(String(255), nullable=True)
    override_model_configs: Mapped[str | None] = mapped_column(LongText)
    conversation_id: Mapped[str] = mapped_column(StringUUID, sa.ForeignKey("conversations.id"), nullable=False)
    _inputs: Mapped[dict[str, Any]] = mapped_column("inputs", sa.JSON)
    query: Mapped[str] = mapped_column(LongText, nullable=False)
    message: Mapped[dict[str, Any]] = mapped_column(sa.JSON, nullable=False)
    message_tokens: Mapped[int] = mapped_column(sa.Integer, nullable=False, server_default=sa.text("0"))
    message_unit_price: Mapped[Decimal] = mapped_column(sa.Numeric(10, 4), nullable=False)
    message_price_unit: Mapped[Decimal] = mapped_column(
        sa.Numeric(10, 7), nullable=False, server_default=sa.text("0.001")
    )
    answer: Mapped[str] = mapped_column(LongText, nullable=False)
    answer_tokens: Mapped[int] = mapped_column(sa.Integer, nullable=False, server_default=sa.text("0"))
    answer_unit_price: Mapped[Decimal] = mapped_column(sa.Numeric(10, 4), nullable=False)
    answer_price_unit: Mapped[Decimal] = mapped_column(
        sa.Numeric(10, 7), nullable=False, server_default=sa.text("0.001")
    )
    parent_message_id: Mapped[str | None] = mapped_column(StringUUID, nullable=True)
    provider_response_latency: Mapped[float] = mapped_column(sa.Float, nullable=False, server_default=sa.text("0"))
    total_price: Mapped[Decimal | None] = mapped_column(sa.Numeric(10, 7))
    currency: Mapped[str] = mapped_column(String(255), nullable=False)
    status: Mapped[MessageStatus] = mapped_column(
        EnumText(MessageStatus, length=255),
        nullable=False,
        server_default=sa.text("'normal'"),
        default=MessageStatus.NORMAL,
    )
    error: Mapped[str | None] = mapped_column(LongText)
    message_metadata: Mapped[str | None] = mapped_column(LongText)
    invoke_from: Mapped[InvokeFrom | None] = mapped_column(EnumText(InvokeFrom, length=255), nullable=True)
    from_source: Mapped[ConversationFromSource] = mapped_column(
        EnumText(ConversationFromSource, length=255), nullable=False
    )
    from_end_user_id: Mapped[str | None] = mapped_column(StringUUID)
    from_account_id: Mapped[str | None] = mapped_column(StringUUID)
    created_at: Mapped[datetime] = mapped_column(sa.DateTime, server_default=func.current_timestamp())
    updated_at: Mapped[datetime] = mapped_column(
        sa.DateTime, nullable=False, server_default=func.current_timestamp(), onupdate=func.current_timestamp()
    )
    agent_based: Mapped[bool] = mapped_column(sa.Boolean, nullable=False, server_default=sa.text("false"))
    workflow_run_id: Mapped[str | None] = mapped_column(StringUUID)
    app_mode: Mapped[AppMode | None] = mapped_column(EnumText(AppMode, length=255), nullable=True)

    @property
    def inputs(self) -> dict[str, Any]:
        inputs = self._inputs.copy()
        # Compatibility bridge: message inputs are persisted as JSON and must remain
        # readable across file payload shape changes. Do not assume `tenant_id`
        # is serialized into each file mapping going forward.
        tenant_resolver = _build_app_tenant_resolver(
            app_id=self.app_id,
            owner_tenant_id=cast(str | None, getattr(self, "_owner_tenant_id", None)),
        )
        for key, value in inputs.items():
            if (
                isinstance(value, dict)
                and cast(dict[str, Any], value).get("dify_model_identity") == FILE_MODEL_IDENTITY
            ):
                value_dict = cast(dict[str, Any], value)
                inputs[key] = build_file_from_input_mapping(
                    file_mapping=value_dict,
                    tenant_resolver=tenant_resolver,
                )
            elif isinstance(value, list):
                value_list = cast(list[Any], value)
                if all(
                    isinstance(item, dict)
                    and cast(dict[str, Any], item).get("dify_model_identity") == FILE_MODEL_IDENTITY
                    for item in value_list
                ):
                    file_list: list[File] = []
                    for item in value_list:
                        if not isinstance(item, dict):
                            continue
                        item_dict = cast(dict[str, Any], item)
                        file_list.append(
                            build_file_from_input_mapping(
                                file_mapping=item_dict,
                                tenant_resolver=tenant_resolver,
                            )
                        )
                    inputs[key] = file_list
        return inputs

    @inputs.setter
    def inputs(self, value: Mapping[str, Any]):
        inputs = dict(value)
        for k, v in inputs.items():
            if isinstance(v, File):
                inputs[k] = v.model_dump()
            elif isinstance(v, list):
                v_list = cast(list[Any], v)
                if all(isinstance(item, File) for item in v_list):
                    inputs[k] = [item.model_dump() for item in v_list if isinstance(item, File)]
        self._inputs = inputs

    @property
    def re_sign_file_url_answer(self) -> str:
        if not self.answer:
            return self.answer

        pattern = r"\[!?.*?\]\((((http|https):\/\/.+)?\/files\/(tools\/)?[\w-]+.*?timestamp=.*&nonce=.*&sign=.*)\)"
        matches = re.findall(pattern, self.answer)

        if not matches:
            return self.answer

        urls = [match[0] for match in matches]

        # remove duplicate urls
        urls = list(set(urls))

        if not urls:
            return self.answer

        re_sign_file_url_answer = self.answer
        for url in urls:
            if "files/tools" in url:
                # get tool file id
                tool_file_id_pattern = r"\/files\/tools\/([\.\w-]+)?\?timestamp="
                result = re.search(tool_file_id_pattern, url)
                if not result:
                    continue

                tool_file_id = result.group(1)

                # get extension
                if "." in tool_file_id:
                    split_result = tool_file_id.split(".")
                    extension = f".{split_result[-1]}"
                    if len(extension) > 10:
                        extension = ".bin"
                    tool_file_id = split_result[0]
                else:
                    extension = ".bin"

                if not tool_file_id:
                    continue

                sign_url = sign_tool_file(tool_file_id=tool_file_id, extension=extension)
            elif "file-preview" in url:
                # get upload file id
                upload_file_id_pattern = r"\/files\/([\w-]+)\/file-preview\?timestamp="
                result = re.search(upload_file_id_pattern, url)
                if not result:
                    continue

                upload_file_id = result.group(1)
                if not upload_file_id:
                    continue
                sign_url = file_helpers.get_signed_file_url(upload_file_id)
            elif "image-preview" in url:
                # image-preview is deprecated, use file-preview instead
                upload_file_id_pattern = r"\/files\/([\w-]+)\/image-preview\?timestamp="
                result = re.search(upload_file_id_pattern, url)
                if not result:
                    continue
                upload_file_id = result.group(1)
                if not upload_file_id:
                    continue
                sign_url = file_helpers.get_signed_file_url(upload_file_id)
            else:
                continue
            # if as_attachment is in the url, add it to the sign_url.
            if "as_attachment" in url:
                sign_url += "&as_attachment=true"
            re_sign_file_url_answer = re_sign_file_url_answer.replace(url, sign_url)

        return re_sign_file_url_answer

    @property
    def user_feedback(self):
        return db.session.scalar(
            select(MessageFeedback).where(MessageFeedback.message_id == self.id, MessageFeedback.from_source == "user")
        )

    @property
    def admin_feedback(self):
        return db.session.scalar(
            select(MessageFeedback).where(MessageFeedback.message_id == self.id, MessageFeedback.from_source == "admin")
        )

    @property
    def feedbacks(self):
        feedbacks = db.session.scalars(select(MessageFeedback).where(MessageFeedback.message_id == self.id)).all()
        return feedbacks

    @property
    def annotation(self):
        annotation = db.session.scalar(select(MessageAnnotation).where(MessageAnnotation.message_id == self.id))
        return annotation

    @property
    def annotation_hit_history(self):
        annotation_history = db.session.scalar(
            select(AppAnnotationHitHistory).where(AppAnnotationHitHistory.message_id == self.id)
        )
        if annotation_history:
            return db.session.scalar(
                select(MessageAnnotation).where(MessageAnnotation.id == annotation_history.annotation_id)
            )
        return None

    @property
    def app_model_config(self):
        conversation = db.session.scalar(select(Conversation).where(Conversation.id == self.conversation_id))
        if conversation:
            return db.session.scalar(
                select(AppModelConfig).where(AppModelConfig.id == conversation.app_model_config_id)
            )

        return None

    @property
    def in_debug_mode(self) -> bool:
        return self.override_model_configs is not None

    @property
    def message_metadata_dict(self) -> dict[str, Any]:
        return json.loads(self.message_metadata) if self.message_metadata else {}

    @property
    def agent_thoughts(self) -> Sequence[MessageAgentThought]:
        return db.session.scalars(
            select(MessageAgentThought)
            .where(MessageAgentThought.message_id == self.id)
            .order_by(MessageAgentThought.position.asc())
        ).all()

    @property
    def retriever_resources(self) -> Any:
        return self.message_metadata_dict.get("retriever_resources") if self.message_metadata else []

    @property
    def message_files(self) -> list[MessageFileInfo]:
        from factories import file_factory

        message_files = db.session.scalars(select(MessageFile).where(MessageFile.message_id == self.id)).all()
        current_app = db.session.scalar(select(App).where(App.id == self.app_id))
        if not current_app:
            raise ValueError(f"App {self.app_id} not found")

        files: list[File] = []
        for message_file in message_files:
            match message_file.transfer_method:
                case FileTransferMethod.LOCAL_FILE:
                    if message_file.upload_file_id is None:
                        raise ValueError(f"MessageFile {message_file.id} is a local file but has no upload_file_id")
                    file = file_factory.build_from_mapping(
                        mapping={
                            "id": message_file.id,
                            "type": message_file.type,
                            "transfer_method": message_file.transfer_method,
                            "upload_file_id": message_file.upload_file_id,
                        },
                        tenant_id=current_app.tenant_id,
                        access_controller=_get_file_access_controller(),
                    )
                case FileTransferMethod.REMOTE_URL:
                    if message_file.url is None:
                        raise ValueError(f"MessageFile {message_file.id} is a remote url but has no url")
                    file = file_factory.build_from_mapping(
                        mapping={
                            "id": message_file.id,
                            "type": message_file.type,
                            "transfer_method": message_file.transfer_method,
                            "upload_file_id": message_file.upload_file_id,
                            "url": message_file.url,
                        },
                        tenant_id=current_app.tenant_id,
                        access_controller=_get_file_access_controller(),
                    )
                case FileTransferMethod.TOOL_FILE:
                    if message_file.upload_file_id is None:
                        assert message_file.url is not None
                        message_file.upload_file_id = message_file.url.split("/")[-1].split(".")[0]
                    mapping = {
                        "id": message_file.id,
                        "type": message_file.type,
                        "transfer_method": message_file.transfer_method,
                        "tool_file_id": message_file.upload_file_id,
                    }
                    file = file_factory.build_from_mapping(
                        mapping=mapping,
                        tenant_id=current_app.tenant_id,
                        access_controller=_get_file_access_controller(),
                    )
                case FileTransferMethod.DATASOURCE_FILE:
                    raise ValueError(
                        f"MessageFile {message_file.id} has an invalid transfer_method {message_file.transfer_method}"
                    )
            files.append(file)

        result = cast(
            list[MessageFileInfo],
            [
                {"belongs_to": message_file.belongs_to, "upload_file_id": message_file.upload_file_id, **file.to_dict()}
                for (file, message_file) in zip(files, message_files)
            ],
        )

        db.session.commit()
        return result

    # TODO(QuantumGhost): dirty hacks, fix this later.
    def set_extra_contents(self, contents: Sequence[dict[str, Any]]) -> None:
        self._extra_contents = list(contents)

    @property
    def extra_contents(self) -> list[ExtraContentDict]:
        return getattr(self, "_extra_contents", [])

    @property
    def workflow_run(self):
        if self.workflow_run_id:
            from sqlalchemy.orm import sessionmaker

            from repositories.factory import DifyAPIRepositoryFactory

            session_maker = sessionmaker(bind=db.engine, expire_on_commit=False)
            repo = DifyAPIRepositoryFactory.create_api_workflow_run_repository(session_maker)
            return repo.get_workflow_run_by_id_without_tenant(run_id=self.workflow_run_id)

        return None

    def to_dict(self) -> MessageDict:
        return {
            "id": self.id,
            "app_id": self.app_id,
            "conversation_id": self.conversation_id,
            "model_id": self.model_id,
            "inputs": self.inputs,
            "query": self.query,
            "total_price": self.total_price,
            "message": self.message,
            "answer": self.answer,
            "status": self.status,
            "error": self.error,
            "message_metadata": self.message_metadata_dict,
            "from_source": self.from_source,
            "from_end_user_id": self.from_end_user_id,
            "from_account_id": self.from_account_id,
            "created_at": self.created_at.isoformat(),
            "updated_at": self.updated_at.isoformat(),
            "agent_based": self.agent_based,
            "workflow_run_id": self.workflow_run_id,
        }

    @classmethod
    def from_dict(cls, data: MessageDict) -> Message:
        return cls(
            id=data["id"],
            app_id=data["app_id"],
            conversation_id=data["conversation_id"],
            model_id=data["model_id"],
            inputs=data["inputs"],
            total_price=data["total_price"],
            query=data["query"],
            message=data["message"],
            answer=data["answer"],
            status=data["status"],
            error=data["error"],
            message_metadata=json.dumps(data["message_metadata"]),
            from_source=data["from_source"],
            from_end_user_id=data["from_end_user_id"],
            from_account_id=data["from_account_id"],
            created_at=data["created_at"],
            updated_at=data["updated_at"],
            agent_based=data["agent_based"],
            workflow_run_id=data["workflow_run_id"],
        )


class MessageFeedback(TypeBase):
    __tablename__ = "message_feedbacks"
    __table_args__ = (
        sa.PrimaryKeyConstraint("id", name="message_feedback_pkey"),
        sa.Index("message_feedback_app_idx", "app_id"),
        sa.Index("message_feedback_message_idx", "message_id", "from_source"),
        sa.Index("message_feedback_conversation_idx", "conversation_id", "from_source", "rating"),
    )

    id: Mapped[str] = mapped_column(
        StringUUID, insert_default=lambda: str(uuid4()), default_factory=lambda: str(uuid4()), init=False
    )
    app_id: Mapped[str] = mapped_column(StringUUID, nullable=False)
    conversation_id: Mapped[str] = mapped_column(StringUUID, nullable=False)
    message_id: Mapped[str] = mapped_column(StringUUID, nullable=False)
    rating: Mapped[FeedbackRating] = mapped_column(EnumText(FeedbackRating, length=255), nullable=False)
    from_source: Mapped[FeedbackFromSource] = mapped_column(EnumText(FeedbackFromSource, length=255), nullable=False)
    content: Mapped[str | None] = mapped_column(LongText, nullable=True, default=None)
    from_end_user_id: Mapped[str | None] = mapped_column(StringUUID, nullable=True, default=None)
    from_account_id: Mapped[str | None] = mapped_column(StringUUID, nullable=True, default=None)
    created_at: Mapped[datetime] = mapped_column(
        sa.DateTime, nullable=False, server_default=func.current_timestamp(), init=False
    )
    updated_at: Mapped[datetime] = mapped_column(
        sa.DateTime,
        nullable=False,
        server_default=func.current_timestamp(),
        onupdate=func.current_timestamp(),
        init=False,
    )

    @property
    def from_account(self) -> Account | None:
        return db.session.scalar(select(Account).where(Account.id == self.from_account_id))

    def to_dict(self) -> MessageFeedbackDict:
        return {
            "id": str(self.id),
            "app_id": str(self.app_id),
            "conversation_id": str(self.conversation_id),
            "message_id": str(self.message_id),
            "rating": self.rating,
            "content": self.content,
            "from_source": self.from_source,
            "from_end_user_id": str(self.from_end_user_id) if self.from_end_user_id else None,
            "from_account_id": str(self.from_account_id) if self.from_account_id else None,
            "created_at": self.created_at.isoformat(),
            "updated_at": self.updated_at.isoformat(),
        }


class MessageFile(TypeBase):
    __tablename__ = "message_files"
    __table_args__ = (
        sa.PrimaryKeyConstraint("id", name="message_file_pkey"),
        sa.Index("message_file_message_idx", "message_id"),
        sa.Index("message_file_created_by_idx", "created_by"),
    )

    id: Mapped[str] = mapped_column(
        StringUUID, insert_default=lambda: str(uuid4()), default_factory=lambda: str(uuid4()), init=False
    )
    message_id: Mapped[str] = mapped_column(StringUUID, nullable=False)
    type: Mapped[FileType] = mapped_column(EnumText(FileType, length=255), nullable=False)
    transfer_method: Mapped[FileTransferMethod] = mapped_column(
        EnumText(FileTransferMethod, length=255), nullable=False
    )
    created_by_role: Mapped[CreatorUserRole] = mapped_column(EnumText(CreatorUserRole, length=255), nullable=False)
    created_by: Mapped[str] = mapped_column(StringUUID, nullable=False)
    belongs_to: Mapped[MessageFileBelongsTo | None] = mapped_column(
        EnumText(MessageFileBelongsTo, length=255), nullable=True, default=None
    )
    url: Mapped[str | None] = mapped_column(LongText, nullable=True, default=None)
    upload_file_id: Mapped[str | None] = mapped_column(StringUUID, nullable=True, default=None)
    created_at: Mapped[datetime] = mapped_column(
        sa.DateTime, nullable=False, server_default=func.current_timestamp(), init=False
    )


class MessageAnnotation(TypeBase):
    __tablename__ = "message_annotations"
    __table_args__ = (
        sa.PrimaryKeyConstraint("id", name="message_annotation_pkey"),
        sa.Index("message_annotation_app_idx", "app_id"),
        sa.Index("message_annotation_conversation_idx", "conversation_id"),
        sa.Index("message_annotation_message_idx", "message_id"),
    )

    id: Mapped[str] = mapped_column(
        StringUUID,
        insert_default=lambda: str(uuid4()),
        default_factory=lambda: str(uuid4()),
        init=False,
    )
    app_id: Mapped[str] = mapped_column(StringUUID)
    question: Mapped[str] = mapped_column(LongText, nullable=False)
    content: Mapped[str] = mapped_column(LongText, nullable=False)
    hit_count: Mapped[int] = mapped_column(sa.Integer, nullable=False, server_default=sa.text("0"), init=False)
    account_id: Mapped[str] = mapped_column(StringUUID, nullable=False)
    conversation_id: Mapped[str | None] = mapped_column(StringUUID, sa.ForeignKey("conversations.id"), default=None)
    message_id: Mapped[str | None] = mapped_column(StringUUID, default=None)
    created_at: Mapped[datetime] = mapped_column(
        sa.DateTime, nullable=False, server_default=func.current_timestamp(), init=False
    )
    updated_at: Mapped[datetime] = mapped_column(
        sa.DateTime,
        nullable=False,
        server_default=func.current_timestamp(),
        onupdate=func.current_timestamp(),
        init=False,
    )

    @property
    def question_text(self) -> str:
        """Return a non-null question string, falling back to the answer content."""
        return self.question or self.content

    @property
    def account(self):
        return db.session.scalar(select(Account).where(Account.id == self.account_id))

    @property
    def annotation_create_account(self):
        return db.session.scalar(select(Account).where(Account.id == self.account_id))


class AppAnnotationHitHistory(TypeBase):
    __tablename__ = "app_annotation_hit_histories"
    __table_args__ = (
        sa.PrimaryKeyConstraint("id", name="app_annotation_hit_histories_pkey"),
        sa.Index("app_annotation_hit_histories_app_idx", "app_id"),
        sa.Index("app_annotation_hit_histories_account_idx", "account_id"),
        sa.Index("app_annotation_hit_histories_annotation_idx", "annotation_id"),
        sa.Index("app_annotation_hit_histories_message_idx", "message_id"),
    )

    id: Mapped[str] = mapped_column(
        StringUUID, insert_default=lambda: str(uuid4()), default_factory=lambda: str(uuid4()), init=False
    )
    app_id: Mapped[str] = mapped_column(StringUUID, nullable=False)
    annotation_id: Mapped[str] = mapped_column(StringUUID, nullable=False)
    source: Mapped[str] = mapped_column(LongText, nullable=False)
    question: Mapped[str] = mapped_column(LongText, nullable=False)
    account_id: Mapped[str] = mapped_column(StringUUID, nullable=False)
    created_at: Mapped[datetime] = mapped_column(
        sa.DateTime, nullable=False, server_default=func.current_timestamp(), init=False
    )
    score: Mapped[float] = mapped_column(Float, nullable=False, server_default=sa.text("0"))
    message_id: Mapped[str] = mapped_column(StringUUID, nullable=False)
    annotation_question: Mapped[str] = mapped_column(LongText, nullable=False)
    annotation_content: Mapped[str] = mapped_column(LongText, nullable=False)

    @property
    def account(self):
        return db.session.scalar(
            select(Account)
            .join(MessageAnnotation, MessageAnnotation.account_id == Account.id)
            .where(MessageAnnotation.id == self.annotation_id)
        )

    @property
    def annotation_create_account(self):
        return db.session.scalar(select(Account).where(Account.id == self.account_id))


class AppAnnotationSetting(TypeBase):
    __tablename__ = "app_annotation_settings"
    __table_args__ = (
        sa.PrimaryKeyConstraint("id", name="app_annotation_settings_pkey"),
        sa.Index("app_annotation_settings_app_idx", "app_id"),
    )

    id: Mapped[str] = mapped_column(
        StringUUID, insert_default=lambda: str(uuid4()), default_factory=lambda: str(uuid4()), init=False
    )
    app_id: Mapped[str] = mapped_column(StringUUID, nullable=False)
    score_threshold: Mapped[float] = mapped_column(Float, nullable=False, server_default=sa.text("0"))
    collection_binding_id: Mapped[str] = mapped_column(StringUUID, nullable=False)
    created_user_id: Mapped[str] = mapped_column(StringUUID, nullable=False)
    created_at: Mapped[datetime] = mapped_column(
        sa.DateTime, nullable=False, server_default=func.current_timestamp(), init=False
    )
    updated_user_id: Mapped[str] = mapped_column(StringUUID, nullable=False)
    updated_at: Mapped[datetime] = mapped_column(
        sa.DateTime,
        nullable=False,
        server_default=func.current_timestamp(),
        onupdate=func.current_timestamp(),
        init=False,
    )

    @property
    def collection_binding_detail(self):
        from .dataset import DatasetCollectionBinding

        return db.session.scalar(
            select(DatasetCollectionBinding).where(DatasetCollectionBinding.id == self.collection_binding_id)
        )


class OperationLog(TypeBase):
    __tablename__ = "operation_logs"
    __table_args__ = (
        sa.PrimaryKeyConstraint("id", name="operation_log_pkey"),
        sa.Index("operation_log_account_action_idx", "tenant_id", "account_id", "action"),
    )

    id: Mapped[str] = mapped_column(
        StringUUID, insert_default=lambda: str(uuid4()), default_factory=lambda: str(uuid4()), init=False
    )
    tenant_id: Mapped[str] = mapped_column(StringUUID, nullable=False)
    account_id: Mapped[str] = mapped_column(StringUUID, nullable=False)
    action: Mapped[str] = mapped_column(String(255), nullable=False)
    content: Mapped[Any | None] = mapped_column(sa.JSON, nullable=True)
    created_at: Mapped[datetime] = mapped_column(
        sa.DateTime, nullable=False, server_default=func.current_timestamp(), init=False
    )
    created_ip: Mapped[str] = mapped_column(String(255), nullable=False)
    updated_at: Mapped[datetime] = mapped_column(
        sa.DateTime,
        nullable=False,
        server_default=func.current_timestamp(),
        onupdate=func.current_timestamp(),
        init=False,
    )


class DefaultEndUserSessionID(StrEnum):
    """
    End User Session ID enum.
    """

    DEFAULT_SESSION_ID = "DEFAULT-USER"


class EndUser(Base, UserMixin):
    __tablename__ = "end_users"
    __table_args__ = (
        sa.PrimaryKeyConstraint("id", name="end_user_pkey"),
        sa.Index("end_user_session_id_idx", "session_id", "type"),
        sa.Index("end_user_tenant_session_id_idx", "tenant_id", "session_id", "type"),
    )

    id: Mapped[str] = mapped_column(StringUUID, default=lambda: str(uuid4()))
    tenant_id: Mapped[str] = mapped_column(StringUUID, nullable=False)
    app_id = mapped_column(StringUUID, nullable=True)
    type: Mapped[str] = mapped_column(String(255), nullable=False)
    external_user_id = mapped_column(String(255), nullable=True)
    name = mapped_column(String(255))
    _is_anonymous: Mapped[bool] = mapped_column(
        "is_anonymous", sa.Boolean, nullable=False, server_default=sa.text("true")
    )

    @property
    def is_anonymous(self) -> Literal[False]:
        return False

    @is_anonymous.setter
    def is_anonymous(self, value: bool) -> None:
        self._is_anonymous = value

    session_id: Mapped[str] = mapped_column(String(255), nullable=False)
    created_at = mapped_column(sa.DateTime, nullable=False, server_default=func.current_timestamp())
    updated_at = mapped_column(
        sa.DateTime, nullable=False, server_default=func.current_timestamp(), onupdate=func.current_timestamp()
    )


class AppMCPServer(TypeBase):
    __tablename__ = "app_mcp_servers"
    __table_args__ = (
        sa.PrimaryKeyConstraint("id", name="app_mcp_server_pkey"),
        sa.UniqueConstraint("tenant_id", "app_id", name="unique_app_mcp_server_tenant_app_id"),
        sa.UniqueConstraint("server_code", name="unique_app_mcp_server_server_code"),
    )
    id: Mapped[str] = mapped_column(
        StringUUID, insert_default=lambda: str(uuid4()), default_factory=lambda: str(uuid4()), init=False
    )
    tenant_id: Mapped[str] = mapped_column(StringUUID, nullable=False)
    app_id: Mapped[str] = mapped_column(StringUUID, nullable=False)
    name: Mapped[str] = mapped_column(String(255), nullable=False)
    description: Mapped[str] = mapped_column(String(255), nullable=False)
    server_code: Mapped[str] = mapped_column(String(255), nullable=False)
    status: Mapped[AppMCPServerStatus] = mapped_column(
        EnumText(AppMCPServerStatus, length=255), nullable=False, server_default=sa.text("'normal'")
    )
    parameters: Mapped[str] = mapped_column(LongText, nullable=False)

    created_at: Mapped[datetime] = mapped_column(
        sa.DateTime, nullable=False, server_default=func.current_timestamp(), init=False
    )
    updated_at: Mapped[datetime] = mapped_column(
        sa.DateTime,
        nullable=False,
        server_default=func.current_timestamp(),
        onupdate=func.current_timestamp(),
        init=False,
    )

    @staticmethod
    def generate_server_code(n: int) -> str:
        while True:
            result = generate_string(n)
            while (
                db.session.scalar(select(func.count(AppMCPServer.id)).where(AppMCPServer.server_code == result)) or 0
            ) > 0:
                result = generate_string(n)

            return result

    @property
    def parameters_dict(self) -> dict[str, str]:
        return cast(dict[str, str], json.loads(self.parameters))


class Site(Base):
    __tablename__ = "sites"
    __table_args__ = (
        sa.PrimaryKeyConstraint("id", name="site_pkey"),
        sa.Index("site_app_id_idx", "app_id"),
        sa.Index("site_code_idx", "code", "status"),
    )

    id = mapped_column(StringUUID, default=lambda: str(uuid4()))
    app_id = mapped_column(StringUUID, nullable=False)
    title: Mapped[str] = mapped_column(String(255), nullable=False)
    icon_type: Mapped[IconType | None] = mapped_column(EnumText(IconType, length=255), nullable=True)
    icon = mapped_column(String(255))
    icon_background = mapped_column(String(255))
    description = mapped_column(LongText)
    default_language: Mapped[str] = mapped_column(String(255), nullable=False)
    chat_color_theme = mapped_column(String(255))
    chat_color_theme_inverted: Mapped[bool] = mapped_column(sa.Boolean, nullable=False, server_default=sa.text("false"))
    copyright = mapped_column(String(255))
    privacy_policy = mapped_column(String(255))
    show_workflow_steps: Mapped[bool] = mapped_column(sa.Boolean, nullable=False, server_default=sa.text("true"))
    use_icon_as_answer_icon: Mapped[bool] = mapped_column(sa.Boolean, nullable=False, server_default=sa.text("false"))
    _custom_disclaimer: Mapped[str] = mapped_column("custom_disclaimer", LongText, default="")
    customize_domain = mapped_column(String(255))
    customize_token_strategy: Mapped[CustomizeTokenStrategy] = mapped_column(
        EnumText(CustomizeTokenStrategy, length=255), nullable=False
    )
    prompt_public: Mapped[bool] = mapped_column(sa.Boolean, nullable=False, server_default=sa.text("false"))
    status: Mapped[AppStatus] = mapped_column(
        EnumText(AppStatus, length=255), nullable=False, server_default=sa.text("'normal'"), default=AppStatus.NORMAL
    )
    created_by = mapped_column(StringUUID, nullable=True)
    created_at = mapped_column(sa.DateTime, nullable=False, server_default=func.current_timestamp())
    updated_by = mapped_column(StringUUID, nullable=True)
    updated_at = mapped_column(
        sa.DateTime, nullable=False, server_default=func.current_timestamp(), onupdate=func.current_timestamp()
    )
    code = mapped_column(String(255))

    @property
    def custom_disclaimer(self):
        return self._custom_disclaimer

    @custom_disclaimer.setter
    def custom_disclaimer(self, value: str):
        if len(value) > 512:
            raise ValueError("Custom disclaimer cannot exceed 512 characters.")
        self._custom_disclaimer = value

    @staticmethod
    def generate_code(n: int) -> str:
        while True:
            result = generate_string(n)
            while (db.session.scalar(select(func.count(Site.id)).where(Site.code == result)) or 0) > 0:
                result = generate_string(n)

            return result

    @property
    def app_base_url(self):
        return dify_config.APP_WEB_URL or request.url_root.rstrip("/")


class ApiToken(Base):  # bug: this uses setattr so idk the field.
    __tablename__ = "api_tokens"
    __table_args__ = (
        sa.PrimaryKeyConstraint("id", name="api_token_pkey"),
        sa.Index("api_token_app_id_type_idx", "app_id", "type"),
        sa.Index("api_token_token_idx", "token", "type"),
        sa.Index("api_token_tenant_idx", "tenant_id", "type"),
    )

    id = mapped_column(StringUUID, default=lambda: str(uuid4()))
    app_id = mapped_column(StringUUID, nullable=True)
    tenant_id = mapped_column(StringUUID, nullable=True)
    type: Mapped[ApiTokenType] = mapped_column(EnumText(ApiTokenType, length=16), nullable=False)
    token: Mapped[str] = mapped_column(String(255), nullable=False)
    last_used_at = mapped_column(sa.DateTime, nullable=True)
    created_at = mapped_column(sa.DateTime, nullable=False, server_default=func.current_timestamp())

    @staticmethod
    def generate_api_key(prefix: str, n: int) -> str:
        while True:
            result = prefix + generate_string(n)
            if db.session.scalar(select(exists().where(ApiToken.token == result))):
                continue
            return result


class UploadFile(TypeBase):
    __tablename__ = "upload_files"
    __table_args__ = (
        sa.PrimaryKeyConstraint("id", name="upload_file_pkey"),
        sa.Index("upload_file_tenant_idx", "tenant_id"),
    )

    # NOTE: The `id` field is generated within the application to minimize extra roundtrips
    # (especially when generating `source_url`) and keep model metadata portable across databases.
    id: Mapped[str] = mapped_column(
        StringUUID,
        init=False,
        default_factory=lambda: str(uuid4()),
    )
    tenant_id: Mapped[str] = mapped_column(StringUUID, nullable=False)
    storage_type: Mapped[StorageType] = mapped_column(EnumText(StorageType, length=255), nullable=False)
    key: Mapped[str] = mapped_column(String(255), nullable=False)
    name: Mapped[str] = mapped_column(String(255), nullable=False)
    size: Mapped[int] = mapped_column(sa.Integer, nullable=False)
    extension: Mapped[str] = mapped_column(String(255), nullable=False)
    mime_type: Mapped[str] = mapped_column(String(255), nullable=True)
    # The `created_by` field stores the ID of the entity that created this upload file.
    #
    # If `created_by_role` is `ACCOUNT`, it corresponds to `Account.id`.
    # Otherwise, it corresponds to `EndUser.id`.
    created_by: Mapped[str] = mapped_column(StringUUID, nullable=False)
    created_at: Mapped[datetime] = mapped_column(sa.DateTime, nullable=False, server_default=func.current_timestamp())

    # The fields `used` and `used_by` are not consistently maintained.
    #
    # When using this model in new code, ensure the following:
    #
    # 1. Set `used` to `true` when the file is utilized.
    # 2. Assign `used_by` to the corresponding `Account.id` or `EndUser.id` based on the `created_by_role`.
    # 3. Avoid relying on these fields for logic, as their values may not always be accurate.
    #
    # `used` may indicate whether the file has been utilized by another service.
    used: Mapped[bool] = mapped_column(sa.Boolean, nullable=False, server_default=sa.text("false"))

    # The `created_by_role` field indicates whether the file was created by an `Account` or an `EndUser`.
    # Its value is derived from the `CreatorUserRole` enumeration.
    created_by_role: Mapped[CreatorUserRole] = mapped_column(
        EnumText(CreatorUserRole, length=255),
        nullable=False,
        server_default=sa.text("'account'"),
        default=CreatorUserRole.ACCOUNT,
    )
    # `used_by` may indicate the ID of the user who utilized this file.
    used_by: Mapped[str | None] = mapped_column(StringUUID, nullable=True, default=None)
    used_at: Mapped[datetime | None] = mapped_column(sa.DateTime, nullable=True, default=None)
    hash: Mapped[str | None] = mapped_column(String(255), nullable=True, default=None)
    source_url: Mapped[str] = mapped_column(LongText, default="")

    def __init__(
        self,
        *,
        tenant_id: str,
        storage_type: StorageType,
        key: str,
        name: str,
        size: int,
        extension: str,
        mime_type: str,
        created_by_role: CreatorUserRole,
        created_by: str,
        created_at: datetime,
        used: bool,
        used_by: str | None = None,
        used_at: datetime | None = None,
        hash: str | None = None,
        source_url: str = "",
    ):
        self.id = str(uuid.uuid4())
        self.tenant_id = tenant_id
        self.storage_type = storage_type
        self.key = key
        self.name = name
        self.size = size
        self.extension = extension
        self.mime_type = mime_type
        self.created_by_role = created_by_role
        self.created_by = created_by
        self.created_at = created_at
        self.used = used
        self.used_by = used_by
        self.used_at = used_at
        self.hash = hash
        self.source_url = source_url


class ApiRequest(TypeBase):
    __tablename__ = "api_requests"
    __table_args__ = (
        sa.PrimaryKeyConstraint("id", name="api_request_pkey"),
        sa.Index("api_request_token_idx", "tenant_id", "api_token_id"),
    )

    id: Mapped[str] = mapped_column(
        StringUUID, insert_default=lambda: str(uuid4()), default_factory=lambda: str(uuid4()), init=False
    )
    tenant_id: Mapped[str] = mapped_column(StringUUID, nullable=False)
    api_token_id: Mapped[str] = mapped_column(StringUUID, nullable=False)
    path: Mapped[str] = mapped_column(String(255), nullable=False)
    request: Mapped[str | None] = mapped_column(LongText, nullable=True)
    response: Mapped[str | None] = mapped_column(LongText, nullable=True)
    ip: Mapped[str] = mapped_column(String(255), nullable=False)
    created_at: Mapped[datetime] = mapped_column(
        sa.DateTime, nullable=False, server_default=func.current_timestamp(), init=False
    )


class MessageChain(TypeBase):
    __tablename__ = "message_chains"
    __table_args__ = (
        sa.PrimaryKeyConstraint("id", name="message_chain_pkey"),
        sa.Index("message_chain_message_id_idx", "message_id"),
    )

    id: Mapped[str] = mapped_column(
        StringUUID, insert_default=lambda: str(uuid4()), default_factory=lambda: str(uuid4()), init=False
    )
    message_id: Mapped[str] = mapped_column(StringUUID, nullable=False)
    type: Mapped[MessageChainType] = mapped_column(EnumText(MessageChainType, length=255), nullable=False)
    input: Mapped[str | None] = mapped_column(LongText, nullable=True)
    output: Mapped[str | None] = mapped_column(LongText, nullable=True)
    created_at: Mapped[datetime] = mapped_column(
        sa.DateTime, nullable=False, server_default=sa.func.current_timestamp(), init=False
    )


class MessageAgentThought(TypeBase):
    __tablename__ = "message_agent_thoughts"
    __table_args__ = (
        sa.PrimaryKeyConstraint("id", name="message_agent_thought_pkey"),
        sa.Index("message_agent_thought_message_id_idx", "message_id"),
        sa.Index("message_agent_thought_message_chain_id_idx", "message_chain_id"),
    )

    id: Mapped[str] = mapped_column(
        StringUUID, insert_default=lambda: str(uuid4()), default_factory=lambda: str(uuid4()), init=False
    )
    message_id: Mapped[str] = mapped_column(StringUUID, nullable=False)
    position: Mapped[int] = mapped_column(sa.Integer, nullable=False)
    created_by_role: Mapped[CreatorUserRole] = mapped_column(EnumText(CreatorUserRole, length=255), nullable=False)
    created_by: Mapped[str] = mapped_column(StringUUID, nullable=False)
    message_chain_id: Mapped[str | None] = mapped_column(StringUUID, nullable=True, default=None)
    thought: Mapped[str | None] = mapped_column(LongText, nullable=True, default=None)
    tool: Mapped[str | None] = mapped_column(LongText, nullable=True, default=None)
    tool_labels_str: Mapped[str] = mapped_column(LongText, nullable=False, default=sa.text("'{}'"))
    tool_meta_str: Mapped[str] = mapped_column(LongText, nullable=False, default=sa.text("'{}'"))
    tool_input: Mapped[str | None] = mapped_column(LongText, nullable=True, default=None)
    observation: Mapped[str | None] = mapped_column(LongText, nullable=True, default=None)
    # plugin_id = mapped_column(StringUUID, nullable=True)  ## for future design
    tool_process_data: Mapped[str | None] = mapped_column(LongText, nullable=True, default=None)
    message: Mapped[str | None] = mapped_column(LongText, nullable=True, default=None)
    message_token: Mapped[int | None] = mapped_column(sa.Integer, nullable=True, default=None)
    message_unit_price: Mapped[Decimal | None] = mapped_column(sa.Numeric, nullable=True, default=None)
    message_price_unit: Mapped[Decimal] = mapped_column(
        sa.Numeric(10, 7), nullable=False, default=Decimal("0.001"), server_default=sa.text("0.001")
    )
    message_files: Mapped[str | None] = mapped_column(LongText, nullable=True, default=None)
    answer: Mapped[str | None] = mapped_column(LongText, nullable=True, default=None)
    answer_token: Mapped[int | None] = mapped_column(sa.Integer, nullable=True, default=None)
    answer_unit_price: Mapped[Decimal | None] = mapped_column(sa.Numeric, nullable=True, default=None)
    answer_price_unit: Mapped[Decimal] = mapped_column(
        sa.Numeric(10, 7), nullable=False, default=Decimal("0.001"), server_default=sa.text("0.001")
    )
    tokens: Mapped[int | None] = mapped_column(sa.Integer, nullable=True, default=None)
    total_price: Mapped[Decimal | None] = mapped_column(sa.Numeric, nullable=True, default=None)
    currency: Mapped[str | None] = mapped_column(String(255), nullable=True, default=None)
    latency: Mapped[float | None] = mapped_column(sa.Float, nullable=True, default=None)
    created_at: Mapped[datetime] = mapped_column(
        sa.DateTime, nullable=False, init=False, server_default=sa.func.current_timestamp()
    )

    @property
    def files(self) -> list[Any]:
        if self.message_files:
            return cast(list[Any], json.loads(self.message_files))
        else:
            return []

    @property
    def tools(self) -> list[str]:
        return self.tool.split(";") if self.tool else []

    @property
    def tool_labels(self) -> dict[str, Any]:
        try:
            if self.tool_labels_str:
                return cast(dict[str, Any], json.loads(self.tool_labels_str))
            else:
                return {}
        except Exception:
            return {}

    @property
    def tool_meta(self) -> dict[str, Any]:
        try:
            if self.tool_meta_str:
                return cast(dict[str, Any], json.loads(self.tool_meta_str))
            else:
                return {}
        except Exception:
            return {}

    @property
    def tool_inputs_dict(self) -> dict[str, Any]:
        tools = self.tools
        try:
            if self.tool_input:
                data = json.loads(self.tool_input)
                result: dict[str, Any] = {}
                for tool in tools:
                    if tool in data:
                        result[tool] = data[tool]
                    else:
                        if len(tools) == 1:
                            result[tool] = data
                        else:
                            result[tool] = {}
                return result
            else:
                return {tool: {} for tool in tools}
        except Exception:
            return {}

    @property
    def tool_outputs_dict(self) -> dict[str, Any]:
        tools = self.tools
        try:
            if self.observation:
                data = json.loads(self.observation)
                result: dict[str, Any] = {}
                for tool in tools:
                    if tool in data:
                        result[tool] = data[tool]
                    else:
                        if len(tools) == 1:
                            result[tool] = data
                        else:
                            result[tool] = {}
                return result
            else:
                return {tool: {} for tool in tools}
        except Exception:
            if self.observation:
                return dict.fromkeys(tools, self.observation)
            else:
                return {}


class DatasetRetrieverResource(TypeBase):
    __tablename__ = "dataset_retriever_resources"
    __table_args__ = (
        sa.PrimaryKeyConstraint("id", name="dataset_retriever_resource_pkey"),
        sa.Index("dataset_retriever_resource_message_id_idx", "message_id"),
    )

    id: Mapped[str] = mapped_column(
        StringUUID, insert_default=lambda: str(uuid4()), default_factory=lambda: str(uuid4()), init=False
    )
    message_id: Mapped[str] = mapped_column(StringUUID, nullable=False)
    position: Mapped[int] = mapped_column(sa.Integer, nullable=False)
    dataset_id: Mapped[str] = mapped_column(StringUUID, nullable=False)
    dataset_name: Mapped[str] = mapped_column(LongText, nullable=False)
    document_id: Mapped[str | None] = mapped_column(StringUUID, nullable=True)
    document_name: Mapped[str] = mapped_column(LongText, nullable=False)
    data_source_type: Mapped[str | None] = mapped_column(LongText, nullable=True)
    segment_id: Mapped[str | None] = mapped_column(StringUUID, nullable=True)
    score: Mapped[float | None] = mapped_column(sa.Float, nullable=True)
    content: Mapped[str] = mapped_column(LongText, nullable=False)
    hit_count: Mapped[int | None] = mapped_column(sa.Integer, nullable=True)
    word_count: Mapped[int | None] = mapped_column(sa.Integer, nullable=True)
    segment_position: Mapped[int | None] = mapped_column(sa.Integer, nullable=True)
    index_node_hash: Mapped[str | None] = mapped_column(LongText, nullable=True)
    retriever_from: Mapped[str] = mapped_column(LongText, nullable=False)
    created_by: Mapped[str] = mapped_column(StringUUID, nullable=False)
    created_at: Mapped[datetime] = mapped_column(
        sa.DateTime, nullable=False, server_default=sa.func.current_timestamp(), init=False
    )


class Tag(TypeBase):
    __tablename__ = "tags"
    __table_args__ = (
        sa.PrimaryKeyConstraint("id", name="tag_pkey"),
        sa.Index("tag_type_idx", "type"),
        sa.Index("tag_name_idx", "name"),
    )

    TAG_TYPE_LIST = ["knowledge", "app"]

    id: Mapped[str] = mapped_column(
        StringUUID, insert_default=lambda: str(uuid4()), default_factory=lambda: str(uuid4()), init=False
    )
    tenant_id: Mapped[str | None] = mapped_column(StringUUID, nullable=True)
    type: Mapped[TagType] = mapped_column(EnumText(TagType, length=16), nullable=False)
    name: Mapped[str] = mapped_column(String(255), nullable=False)
    created_by: Mapped[str] = mapped_column(StringUUID, nullable=False)
    created_at: Mapped[datetime] = mapped_column(
        sa.DateTime, nullable=False, server_default=func.current_timestamp(), init=False
    )


class TagBinding(TypeBase):
    __tablename__ = "tag_bindings"
    __table_args__ = (
        sa.PrimaryKeyConstraint("id", name="tag_binding_pkey"),
        sa.Index("tag_bind_target_id_idx", "target_id"),
        sa.Index("tag_bind_tag_id_idx", "tag_id"),
    )

    id: Mapped[str] = mapped_column(
        StringUUID, insert_default=lambda: str(uuid4()), default_factory=lambda: str(uuid4()), init=False
    )
    tenant_id: Mapped[str | None] = mapped_column(StringUUID, nullable=True)
    tag_id: Mapped[str | None] = mapped_column(StringUUID, nullable=True)
    target_id: Mapped[str | None] = mapped_column(StringUUID, nullable=True)
    created_by: Mapped[str] = mapped_column(StringUUID, nullable=False)
    created_at: Mapped[datetime] = mapped_column(
        sa.DateTime, nullable=False, server_default=func.current_timestamp(), init=False
    )


class TraceAppConfig(TypeBase):
    __tablename__ = "trace_app_config"
    __table_args__ = (
        sa.PrimaryKeyConstraint("id", name="tracing_app_config_pkey"),
        sa.Index("trace_app_config_app_id_idx", "app_id"),
    )

    id: Mapped[str] = mapped_column(
        StringUUID, insert_default=lambda: str(uuid4()), default_factory=lambda: str(uuid4()), init=False
    )
    app_id: Mapped[str] = mapped_column(StringUUID, nullable=False)
    tracing_provider: Mapped[str | None] = mapped_column(String(255), nullable=True)
    tracing_config: Mapped[dict[str, Any] | None] = mapped_column(sa.JSON, nullable=True)
    created_at: Mapped[datetime] = mapped_column(
        sa.DateTime, nullable=False, server_default=func.current_timestamp(), init=False
    )
    updated_at: Mapped[datetime] = mapped_column(
        sa.DateTime,
        nullable=False,
        server_default=func.current_timestamp(),
        onupdate=func.current_timestamp(),
        init=False,
    )
    is_active: Mapped[bool] = mapped_column(sa.Boolean, nullable=False, server_default=sa.text("true"), default=True)

    @property
    def tracing_config_dict(self) -> dict[str, Any]:
        return self.tracing_config or {}

    @property
    def tracing_config_str(self) -> str:
        return json.dumps(self.tracing_config_dict)

    def to_dict(self) -> TraceAppConfigDict:
        return {
            "id": self.id,
            "app_id": self.app_id,
            "tracing_provider": self.tracing_provider,
            "tracing_config": self.tracing_config_dict,
            "is_active": self.is_active,
            "created_at": str(self.created_at) if self.created_at else None,
            "updated_at": str(self.updated_at) if self.updated_at else None,
        }


class TenantCreditPool(TypeBase):
    __tablename__ = "tenant_credit_pools"
    __table_args__ = (
        sa.PrimaryKeyConstraint("id", name="tenant_credit_pool_pkey"),
        sa.Index("tenant_credit_pool_tenant_id_idx", "tenant_id"),
        sa.Index("tenant_credit_pool_pool_type_idx", "pool_type"),
    )

    id: Mapped[str] = mapped_column(
        StringUUID, insert_default=lambda: str(uuid4()), default_factory=lambda: str(uuid4()), init=False
    )
    tenant_id: Mapped[str] = mapped_column(StringUUID, nullable=False)
    pool_type: Mapped[ProviderQuotaType] = mapped_column(
        EnumText(ProviderQuotaType, length=40), nullable=False, default=ProviderQuotaType.TRIAL, server_default="trial"
    )
    quota_limit: Mapped[int] = mapped_column(BigInteger, nullable=False, default=0)
    quota_used: Mapped[int] = mapped_column(BigInteger, nullable=False, default=0)
    created_at: Mapped[datetime] = mapped_column(
        sa.DateTime, nullable=False, server_default=text("CURRENT_TIMESTAMP"), init=False
    )
    updated_at: Mapped[datetime] = mapped_column(
        sa.DateTime,
        nullable=False,
        server_default=func.current_timestamp(),
        onupdate=func.current_timestamp(),
        init=False,
    )

    @property
    def remaining_credits(self) -> int:
        return max(0, self.quota_limit - self.quota_used)

    def has_sufficient_credits(self, required_credits: int) -> bool:
        return self.remaining_credits >= required_credits