fix: template

Co-authored-by: Copilot <175728472+Copilot@users.noreply.github.com>
Co-authored-by: gemini-code-assist[bot] <176961590+gemini-code-assist[bot]@users.noreply.github.com>

.github/workflows/api-tests.yml

@@ -62,7 +62,7 @@ jobs:
           compose-file: |
             docker/docker-compose.middleware.yaml
           services: |
-            db
+            db_postgres
             redis
             sandbox
             ssrf_proxy

.github/workflows/db-migration-test.yml

@@ -8,7 +8,7 @@ concurrency:
   cancel-in-progress: true
 
 jobs:
-  db-migration-test:
+  db-migration-test-postgres:
     runs-on: ubuntu-latest
 
     steps:
@@ -45,7 +45,7 @@ jobs:
           compose-file: |
             docker/docker-compose.middleware.yaml
           services: |
-            db
+            db_postgres
             redis
 
       - name: Prepare configs
@@ -57,3 +57,54 @@ jobs:
         env:
           DEBUG: true
         run: uv run --directory api flask upgrade-db
+
+  db-migration-test-mysql:
+    runs-on: ubuntu-latest
+
+    steps:
+      - name: Checkout code
+        uses: actions/checkout@v4
+        with:
+          fetch-depth: 0
+          persist-credentials: false
+
+      - name: Setup UV and Python
+        uses: astral-sh/setup-uv@v6
+        with:
+          enable-cache: true
+          python-version: "3.12"
+          cache-dependency-glob: api/uv.lock
+
+      - name: Install dependencies
+        run: uv sync --project api
+      - name: Ensure Offline migration are supported
+        run: |
+          # upgrade
+          uv run --directory api flask db upgrade 'base:head' --sql
+          # downgrade
+          uv run --directory api flask db downgrade 'head:base' --sql
+
+      - name: Prepare middleware env
+        run: |
+          cd docker
+          cp middleware.env.example middleware.env
+
+      - name: Set up Middlewares
+        uses: hoverkraft-tech/compose-action@v2.0.2
+        with:
+          compose-file: |
+            docker/docker-compose.middleware.yaml
+          services: |
+            db_mysql
+            redis
+
+      - name: Prepare configs for MySQL
+        run: |
+          cd api
+          cp .env.example .env
+          sed -i 's/DB_TYPE=postgresql/DB_TYPE=mysql/' .env
+
+      - name: Run DB Migration
+        env:
+          DEBUG: true
+        run: uv run --directory api flask upgrade-db

.gitignore

@@ -183,6 +183,7 @@ docker/volumes/couchbase/*
 docker/volumes/oceanbase/*
 docker/volumes/plugin_daemon/*
 docker/volumes/matrixone/*
+docker/volumes/mysql/data/*
 !docker/volumes/oceanbase/init.d
 
 docker/nginx/conf.d/default.conf

README.md

@@ -117,7 +117,7 @@ All of Dify's offerings come with corresponding APIs, so you could effortlessly
   Use our [documentation](https://docs.dify.ai) for further references and more in-depth instructions.
 
 - **Dify for enterprise / organizations<br/>**
-  We provide additional enterprise-centric features. [Log your questions for us through this chatbot](https://udify.app/chat/22L1zSxg6yW1cWQg) or [send us an email](mailto:business@dify.ai?subject=%5BGitHub%5DBusiness%20License%20Inquiry) to discuss enterprise needs. <br/>
+  We provide additional enterprise-centric features. [Send us an email](mailto:business@dify.ai?subject=%5BGitHub%5DBusiness%20License%20Inquiry) to discuss your enterprise needs. <br/>
 
   > For startups and small businesses using AWS, check out [Dify Premium on AWS Marketplace](https://aws.amazon.com/marketplace/pp/prodview-t22mebxzwjhu6) and deploy it to your own AWS VPC with one click. It's an affordable AMI offering with the option to create apps with custom logo and branding.
 

api/.env.example

@@ -30,9 +30,6 @@ INTERNAL_FILES_URL=http://127.0.0.1:5001
 # The time in seconds after the signature is rejected
 FILES_ACCESS_TIMEOUT=300
 
-# Collaboration mode toggle
-ENABLE_COLLABORATION_MODE=false
-
 # Access token expiration time in minutes
 ACCESS_TOKEN_EXPIRE_MINUTES=60
 
@@ -72,12 +69,31 @@ REDIS_CLUSTERS_PASSWORD=
 # celery configuration
 CELERY_BROKER_URL=redis://:difyai123456@localhost:${REDIS_PORT}/1
 CELERY_BACKEND=redis
-# PostgreSQL database configuration
-DB_USERNAME=postgres
-DB_PASSWORD=difyai123456
-DB_HOST=localhost
-DB_PORT=5432
-DB_DATABASE=dify
+
+# Database configuration
+DB_TYPE=postgresql
+
+# PostgreSQL configuration
+POSTGRES_USER=postgres
+POSTGRES_PASSWORD=difyai123456
+POSTGRES_HOST=localhost
+POSTGRES_PORT=5432
+POSTGRES_DATABASE=dify
+
+# MySQL configuration
+MYSQL_USER=mysql
+MYSQL_PASSWORD=difyai123456
+MYSQL_HOST=localhost
+MYSQL_PORT=3306
+MYSQL_DATABASE=dify
+
+# OceanBase configuration
+OCEANBASE_USER=root@test
+OCEANBASE_PASSWORD=difyai123456
+OCEANBASE_HOST=localhost
+OCEANBASE_PORT=2881
+OCEANBASE_DATABASE=test
+
 SQLALCHEMY_POOL_PRE_PING=true
 SQLALCHEMY_POOL_TIMEOUT=30
 
@@ -175,6 +191,15 @@ WEAVIATE_API_KEY=WVF5YThaHlkYwhGUSmCRgsX3tD5ngdN8pkih
 WEAVIATE_GRPC_ENABLED=false
 WEAVIATE_BATCH_SIZE=100
 
+# OceanBase Vector configuration
+OCEANBASE_HOST=127.0.0.1
+OCEANBASE_PORT=2881
+OCEANBASE_USER=root@test
+OCEANBASE_PASSWORD=difyai123456
+OCEANBASE_DATABASE=test
+OCEANBASE_MEMORY_LIMIT=6G
+OCEANBASE_ENABLE_HYBRID_SEARCH=false
+
 # Qdrant configuration, use `http://localhost:6333` for local mode or `https://your-qdrant-cluster-url.qdrant.io` for remote mode
 QDRANT_URL=http://localhost:6333
 QDRANT_API_KEY=difyai123456
@@ -340,15 +365,6 @@ LINDORM_PASSWORD=admin
 LINDORM_USING_UGC=True
 LINDORM_QUERY_TIMEOUT=1
 
-# OceanBase Vector configuration
-OCEANBASE_VECTOR_HOST=127.0.0.1
-OCEANBASE_VECTOR_PORT=2881
-OCEANBASE_VECTOR_USER=root@test
-OCEANBASE_VECTOR_PASSWORD=difyai123456
-OCEANBASE_VECTOR_DATABASE=test
-OCEANBASE_MEMORY_LIMIT=6G
-OCEANBASE_ENABLE_HYBRID_SEARCH=false
-
 # AlibabaCloud MySQL Vector configuration
 ALIBABACLOUD_MYSQL_HOST=127.0.0.1
 ALIBABACLOUD_MYSQL_PORT=3306
@@ -618,5 +634,8 @@ SWAGGER_UI_PATH=/swagger-ui.html
 # Set to false to export dataset IDs as plain text for easier cross-environment import
 DSL_EXPORT_ENCRYPT_DATASET_ID=true
 
+# Tenant isolated task queue configuration
+TENANT_ISOLATED_TASK_CONCURRENCY=1
+
 # Maximum number of segments for dataset segments API (0 for unlimited)
 DATASET_MAX_SEGMENTS_PER_REQUEST=0

api/README.md

@@ -15,8 +15,8 @@
    ```bash
    cd ../docker
    cp middleware.env.example middleware.env
-   # change the profile to other vector database if you are not using weaviate
-   docker compose -f docker-compose.middleware.yaml --profile weaviate -p dify up -d
+   # change the profile to mysql if you are not using postgres,change the profile to other vector database if you are not using weaviate
+   docker compose -f docker-compose.middleware.yaml --profile postgresql --profile weaviate -p dify up -d
    cd ../api
    ```
 

api/app.py

@@ -1,4 +1,3 @@
-import os
 import sys
 
 
@@ -9,16 +8,10 @@ def is_db_command():
 
 
 # create app
-celery = None
-flask_app = None
-socketio_app = None
-
 if is_db_command():
     from app_factory import create_migrations_app
 
     app = create_migrations_app()
-    socketio_app = app
-    flask_app = app
 else:
     # Gunicorn and Celery handle monkey patching automatically in production by
     # specifying the `gevent` worker class. Manual monkey patching is not required here.
@@ -29,15 +22,8 @@ else:
 
     from app_factory import create_app
 
-    socketio_app, flask_app = create_app()
-    app = flask_app
-    celery = flask_app.extensions["celery"]
+    app = create_app()
+    celery = app.extensions["celery"]
 
 if __name__ == "__main__":
-    from gevent import pywsgi
-    from geventwebsocket.handler import WebSocketHandler
-
-    host = os.environ.get("HOST", "0.0.0.0")
-    port = int(os.environ.get("PORT", 5001))
-    server = pywsgi.WSGIServer((host, port), socketio_app, handler_class=WebSocketHandler)
-    server.serve_forever()
+    app.run(host="0.0.0.0", port=5001)

api/app_factory.py

@@ -31,22 +31,14 @@ def create_flask_app_with_configs() -> DifyApp:
     return dify_app
 
 
-def create_app() -> tuple[any, DifyApp]:
+def create_app() -> DifyApp:
     start_time = time.perf_counter()
     app = create_flask_app_with_configs()
     initialize_extensions(app)
-
-    import socketio
-
-    from extensions.ext_socketio import sio
-
-    sio.app = app
-    socketio_app = socketio.WSGIApp(sio, app)
-
     end_time = time.perf_counter()
     if dify_config.DEBUG:
         logger.info("Finished create_app (%s ms)", round((end_time - start_time) * 1000, 2))
-    return socketio_app, app
+    return app
 
 
 def initialize_extensions(app: DifyApp):

api/configs/feature/__init__.py

@@ -871,16 +871,6 @@ class MailConfig(BaseSettings):
         default=None,
     )
 
-    ENABLE_TRIAL_APP: bool = Field(
-        description="Enable trial app",
-        default=False,
-    )
-
-    ENABLE_EXPLORE_BANNER: bool = Field(
-        description="Enable explore banner",
-        default=False,
-    )
-
 
 class RagEtlConfig(BaseSettings):
     """
@@ -1093,13 +1083,6 @@ class PositionConfig(BaseSettings):
         return {item.strip() for item in self.POSITION_TOOL_EXCLUDES.split(",") if item.strip() != ""}
 
 
-class CollaborationConfig(BaseSettings):
-    ENABLE_COLLABORATION_MODE: bool = Field(
-        description="Whether to enable collaboration mode features across the workspace",
-        default=False,
-    )
-
-
 class LoginConfig(BaseSettings):
     ENABLE_EMAIL_CODE_LOGIN: bool = Field(
         description="whether to enable email code login",
@@ -1159,9 +1142,9 @@ class SwaggerUIConfig(BaseSettings):
     )
 
 
-class TenantSelfTaskQueueConfig(BaseSettings):
-    TENANT_SELF_TASK_QUEUE_PULL_SIZE: int = Field(
-        description="Default batch size for tenant self task queue pull operations",
+class TenantIsolatedTaskQueueConfig(BaseSettings):
+    TENANT_ISOLATED_TASK_CONCURRENCY: int = Field(
+        description="Number of tasks allowed to be delivered concurrently from isolated queue per tenant",
         default=1,
     )
 
@@ -1190,13 +1173,12 @@ class FeatureConfig(
     RagEtlConfig,
     RepositoryConfig,
     SecurityConfig,
-    TenantSelfTaskQueueConfig,
+    TenantIsolatedTaskQueueConfig,
     ToolConfig,
     UpdateConfig,
     WorkflowConfig,
     WorkflowNodeExecutionConfig,
     WorkspaceConfig,
-    CollaborationConfig,
     LoginConfig,
     AccountConfig,
     SwaggerUIConfig,

api/configs/feature/hosted_service/__init__.py

@@ -8,11 +8,6 @@ class HostedCreditConfig(BaseSettings):
         default="",
     )
 
-    HOSTED_POOL_CREDITS: int = Field(
-        description="Pool credits for hosted service",
-        default=200,
-    )
-
     def get_model_credits(self, model_name: str) -> int:
         """
         Get credit value for a specific model name.
@@ -65,46 +60,19 @@ class HostedOpenAiConfig(BaseSettings):
 
     HOSTED_OPENAI_TRIAL_MODELS: str = Field(
         description="Comma-separated list of available models for trial access",
-        default="gpt-4,"
-        "gpt-4-turbo-preview,"
-        "gpt-4-turbo-2024-04-09,"
-        "gpt-4-1106-preview,"
-        "gpt-4-0125-preview,"
-        "gpt-4-turbo,"
-        "gpt-4.1,"
-        "gpt-4.1-2025-04-14,"
-        "gpt-4.1-mini,"
-        "gpt-4.1-mini-2025-04-14,"
-        "gpt-4.1-nano,"
-        "gpt-4.1-nano-2025-04-14,"
-        "gpt-3.5-turbo,"
+        default="gpt-3.5-turbo,"
+        "gpt-3.5-turbo-1106,"
+        "gpt-3.5-turbo-instruct,"
         "gpt-3.5-turbo-16k,"
         "gpt-3.5-turbo-16k-0613,"
-        "gpt-3.5-turbo-1106,"
         "gpt-3.5-turbo-0613,"
         "gpt-3.5-turbo-0125,"
-        "gpt-3.5-turbo-instruct,"
-        "text-davinci-003,"
-        "chatgpt-4o-latest,"
-        "gpt-4o,"
-        "gpt-4o-2024-05-13,"
-        "gpt-4o-2024-08-06,"
-        "gpt-4o-2024-11-20,"
-        "gpt-4o-audio-preview,"
-        "gpt-4o-audio-preview-2025-06-03,"
-        "gpt-4o-mini,"
-        "gpt-4o-mini-2024-07-18,"
-        "o3-mini,"
-        "o3-mini-2025-01-31,"
-        "gpt-5-mini-2025-08-07,"
-        "gpt-5-mini,"
-        "o4-mini,"
-        "o4-mini-2025-04-16,"
-        "gpt-5-chat-latest,"
-        "gpt-5,"
-        "gpt-5-2025-08-07,"
-        "gpt-5-nano,"
-        "gpt-5-nano-2025-08-07",
+        "text-davinci-003",
+    )
+
+    HOSTED_OPENAI_QUOTA_LIMIT: NonNegativeInt = Field(
+        description="Quota limit for hosted OpenAI service usage",
+        default=200,
     )
 
     HOSTED_OPENAI_PAID_ENABLED: bool = Field(
@@ -119,13 +87,6 @@ class HostedOpenAiConfig(BaseSettings):
         "gpt-4-turbo-2024-04-09,"
         "gpt-4-1106-preview,"
         "gpt-4-0125-preview,"
-        "gpt-4-turbo,"
-        "gpt-4.1,"
-        "gpt-4.1-2025-04-14,"
-        "gpt-4.1-mini,"
-        "gpt-4.1-mini-2025-04-14,"
-        "gpt-4.1-nano,"
-        "gpt-4.1-nano-2025-04-14,"
         "gpt-3.5-turbo,"
         "gpt-3.5-turbo-16k,"
         "gpt-3.5-turbo-16k-0613,"
@@ -133,150 +94,7 @@ class HostedOpenAiConfig(BaseSettings):
         "gpt-3.5-turbo-0613,"
         "gpt-3.5-turbo-0125,"
         "gpt-3.5-turbo-instruct,"
-        "text-davinci-003,"
-        "chatgpt-4o-latest,"
-        "gpt-4o,"
-        "gpt-4o-2024-05-13,"
-        "gpt-4o-2024-08-06,"
-        "gpt-4o-2024-11-20,"
-        "gpt-4o-audio-preview,"
-        "gpt-4o-audio-preview-2025-06-03,"
-        "gpt-4o-mini,"
-        "gpt-4o-mini-2024-07-18,"
-        "o3-mini,"
-        "o3-mini-2025-01-31,"
-        "gpt-5-mini-2025-08-07,"
-        "gpt-5-mini,"
-        "o4-mini,"
-        "o4-mini-2025-04-16,"
-        "gpt-5-chat-latest,"
-        "gpt-5,"
-        "gpt-5-2025-08-07,"
-        "gpt-5-nano,"
-        "gpt-5-nano-2025-08-07",
-    )
-
-
-class HostedGeminiConfig(BaseSettings):
-    """
-    Configuration for fetching Gemini service
-    """
-
-    HOSTED_GEMINI_API_KEY: str | None = Field(
-        description="API key for hosted Gemini service",
-        default=None,
-    )
-
-    HOSTED_GEMINI_API_BASE: str | None = Field(
-        description="Base URL for hosted Gemini API",
-        default=None,
-    )
-
-    HOSTED_GEMINI_API_ORGANIZATION: str | None = Field(
-        description="Organization ID for hosted Gemini service",
-        default=None,
-    )
-
-    HOSTED_GEMINI_TRIAL_ENABLED: bool = Field(
-        description="Enable trial access to hosted Gemini service",
-        default=False,
-    )
-
-    HOSTED_GEMINI_TRIAL_MODELS: str = Field(
-        description="Comma-separated list of available models for trial access",
-        default="gemini-2.5-flash,gemini-2.0-flash,gemini-2.0-flash-lite,",
-    )
-
-    HOSTED_GEMINI_PAID_ENABLED: bool = Field(
-        description="Enable paid access to hosted gemini service",
-        default=False,
-    )
-
-    HOSTED_GEMINI_PAID_MODELS: str = Field(
-        description="Comma-separated list of available models for paid access",
-        default="gemini-2.5-flash,gemini-2.0-flash,gemini-2.0-flash-lite,",
-    )
-
-
-class HostedXAIConfig(BaseSettings):
-    """
-    Configuration for fetching XAI service
-    """
-
-    HOSTED_XAI_API_KEY: str | None = Field(
-        description="API key for hosted XAI service",
-        default=None,
-    )
-
-    HOSTED_XAI_API_BASE: str | None = Field(
-        description="Base URL for hosted XAI API",
-        default=None,
-    )
-
-    HOSTED_XAI_API_ORGANIZATION: str | None = Field(
-        description="Organization ID for hosted XAI service",
-        default=None,
-    )
-
-    HOSTED_XAI_TRIAL_ENABLED: bool = Field(
-        description="Enable trial access to hosted XAI service",
-        default=False,
-    )
-
-    HOSTED_XAI_TRIAL_MODELS: str = Field(
-        description="Comma-separated list of available models for trial access",
-        default="grok-3,grok-3-mini,grok-3-mini-fast",
-    )
-
-    HOSTED_XAI_PAID_ENABLED: bool = Field(
-        description="Enable paid access to hosted XAI service",
-        default=False,
-    )
-
-    HOSTED_XAI_PAID_MODELS: str = Field(
-        description="Comma-separated list of available models for paid access",
-        default="grok-3,grok-3-mini,grok-3-mini-fast",
-    )
-
-
-class HostedDeepseekConfig(BaseSettings):
-    """
-    Configuration for fetching Deepseek service
-    """
-
-    HOSTED_DEEPSEEK_API_KEY: str | None = Field(
-        description="API key for hosted Deepseek service",
-        default=None,
-    )
-
-    HOSTED_DEEPSEEK_API_BASE: str | None = Field(
-        description="Base URL for hosted Deepseek API",
-        default=None,
-    )
-
-    HOSTED_DEEPSEEK_API_ORGANIZATION: str | None = Field(
-        description="Organization ID for hosted Deepseek service",
-        default=None,
-    )
-
-    HOSTED_DEEPSEEK_TRIAL_ENABLED: bool = Field(
-        description="Enable trial access to hosted Deepseek service",
-        default=False,
-    )
-
-    HOSTED_DEEPSEEK_TRIAL_MODELS: str = Field(
-        description="Comma-separated list of available models for trial access",
-        default="deepseek-chat,deepseek-reasoner",
-    )
-
-    HOSTED_DEEPSEEK_PAID_ENABLED: bool = Field(
-        description="Enable paid access to hosted XAI service",
-        default=False,
-    )
-
-    HOSTED_DEEPSEEK_PAID_MODELS: str = Field(
-        description="Comma-separated list of available models for paid access",
-        default="deepseek-chat,deepseek-reasoner",
+        "text-davinci-003",
     )
 
 
@@ -326,30 +144,16 @@ class HostedAnthropicConfig(BaseSettings):
         default=False,
     )
 
+    HOSTED_ANTHROPIC_QUOTA_LIMIT: NonNegativeInt = Field(
+        description="Quota limit for hosted Anthropic service usage",
+        default=600000,
+    )
+
     HOSTED_ANTHROPIC_PAID_ENABLED: bool = Field(
         description="Enable paid access to hosted Anthropic service",
         default=False,
     )
 
-    HOSTED_ANTHROPIC_TRIAL_MODELS: str = Field(
-        description="Comma-separated list of available models for paid access",
-        default="claude-opus-4-20250514,"
-        "claude-sonnet-4-20250514,"
-        "claude-3-5-haiku-20241022,"
-        "claude-3-opus-20240229,"
-        "claude-3-7-sonnet-20250219,"
-        "claude-3-haiku-20240307",
-    )
-    HOSTED_ANTHROPIC_PAID_MODELS: str = Field(
-        description="Comma-separated list of available models for paid access",
-        default="claude-opus-4-20250514,"
-        "claude-sonnet-4-20250514,"
-        "claude-3-5-haiku-20241022,"
-        "claude-3-opus-20240229,"
-        "claude-3-7-sonnet-20250219,"
-        "claude-3-haiku-20240307",
-    )
-
 
 class HostedMinmaxConfig(BaseSettings):
     """
@@ -446,8 +250,5 @@ class HostedServiceConfig(
     HostedModerationConfig,
     # credit config
     HostedCreditConfig,
-    HostedGeminiConfig,
-    HostedXAIConfig,
-    HostedDeepseekConfig,
 ):
     pass

api/configs/middleware/__init__.py

@@ -105,31 +105,141 @@ class KeywordStoreConfig(BaseSettings):
 
 
 class DatabaseConfig(BaseSettings):
-    DB_HOST: str = Field(
-        description="Hostname or IP address of the database server.",
+    # Database type selector
+    DB_TYPE: Literal["postgresql", "mysql", "oceanbase"] = Field(
+        description="Database type to use. OceanBase is MySQL-compatible.",
+        default="postgresql",
+    )
+    
+    # PostgreSQL configuration
+    POSTGRES_HOST: str = Field(
+        description="PostgreSQL hostname or IP address.",
         default="localhost",
     )
 
-    DB_PORT: PositiveInt = Field(
-        description="Port number for database connection.",
+    POSTGRES_PORT: PositiveInt = Field(
+        description="PostgreSQL port number.",
         default=5432,
     )
 
-    DB_USERNAME: str = Field(
-        description="Username for database authentication.",
+    POSTGRES_USER: str = Field(
+        description="PostgreSQL username.",
         default="postgres",
     )
 
-    DB_PASSWORD: str = Field(
-        description="Password for database authentication.",
-        default="",
+    POSTGRES_PASSWORD: str = Field(
+        description="PostgreSQL password.",
+        default="difyai123456",
     )
 
-    DB_DATABASE: str = Field(
-        description="Name of the database to connect to.",
+    POSTGRES_DATABASE: str = Field(
+        description="PostgreSQL database name.",
         default="dify",
     )
 
+    # MySQL configuration
+    MYSQL_HOST: str = Field(
+        description="MySQL hostname or IP address.",
+        default="localhost",
+    )
+    
+    MYSQL_PORT: PositiveInt = Field(
+        description="MySQL port number.",
+        default=3306,
+    )
+
+    MYSQL_USER: str = Field(
+        description="MySQL username.",
+        default="root",
+    )
+
+    MYSQL_PASSWORD: str = Field(
+        description="MySQL password.",
+        default="difyai123456",
+    )
+
+    MYSQL_DATABASE: str = Field(
+        description="MySQL database name.",
+        default="dify",
+    )
+
+    # OceanBase configuration(MySQL-compatible)
+    OCEANBASE_HOST: str = Field(
+        description="OceanBase hostname or IP address.",
+        default="localhost",
+    )
+    
+    OCEANBASE_PORT: PositiveInt = Field(
+        description="OceanBase port number.",
+        default=2881,
+    )
+    
+    OCEANBASE_USER: str = Field(
+        description="OceanBase username.",
+        default="root@test",
+    )
+    
+    OCEANBASE_PASSWORD: str = Field(
+        description="OceanBase password.",
+        default="difyai123456",
+    )
+    
+    OCEANBASE_DATABASE: str = Field(
+        description="OceanBase database name.",
+        default="test",
+    )
+
+    # Dynamic properties based on DB_TYPE
+    @computed_field  # type: ignore[prop-decorator]
+    @property
+    def DB_HOST(self) -> str:
+        if self.DB_TYPE == "postgresql":
+            return self.POSTGRES_HOST
+        elif self.DB_TYPE == "mysql":
+            return self.MYSQL_HOST
+        elif self.DB_TYPE == "oceanbase":
+            return self.OCEANBASE_HOST
+
+    @computed_field  # type: ignore[prop-decorator]
+    @property
+    def DB_PORT(self) -> int:
+        if self.DB_TYPE == "postgresql":
+            return self.POSTGRES_PORT
+        elif self.DB_TYPE == "mysql":
+            return self.MYSQL_PORT
+        elif self.DB_TYPE == "oceanbase":
+            return self.OCEANBASE_PORT
+    
+    @computed_field  # type: ignore[prop-decorator]
+    @property
+    def DB_USERNAME(self) -> str:
+        if self.DB_TYPE == "postgresql":
+            return self.POSTGRES_USER
+        elif self.DB_TYPE == "mysql":
+            return self.MYSQL_USER
+        elif self.DB_TYPE == "oceanbase":
+            return self.OCEANBASE_USER
+    
+    @computed_field  # type: ignore[prop-decorator]
+    @property
+    def DB_PASSWORD(self) -> str:
+        if self.DB_TYPE == "postgresql":
+            return self.POSTGRES_PASSWORD
+        elif self.DB_TYPE == "mysql":
+            return self.MYSQL_PASSWORD
+        elif self.DB_TYPE == "oceanbase":
+            return self.OCEANBASE_PASSWORD
+
+    @computed_field  # type: ignore[prop-decorator]
+    @property
+    def DB_DATABASE(self) -> str:
+        if self.DB_TYPE == "postgresql":
+            return self.POSTGRES_DATABASE
+        elif self.DB_TYPE == "mysql":
+            return self.MYSQL_DATABASE
+        elif self.DB_TYPE == "oceanbase":
+            return self.OCEANBASE_DATABASE
+
     DB_CHARSET: str = Field(
         description="Character set for database connection.",
         default="",
@@ -140,10 +250,10 @@ class DatabaseConfig(BaseSettings):
         default="",
     )
 
-    SQLALCHEMY_DATABASE_URI_SCHEME: str = Field(
-        description="Database URI scheme for SQLAlchemy connection.",
-        default="postgresql",
-    )
+    @computed_field  # type: ignore[prop-decorator]
+    @property
+    def SQLALCHEMY_DATABASE_URI_SCHEME(self) -> str:
+        return "postgresql" if self.DB_TYPE == "postgresql" else "mysql+pymysql"
 
     @computed_field  # type: ignore[prop-decorator]
     @property
@@ -204,15 +314,15 @@ class DatabaseConfig(BaseSettings):
         # Parse DB_EXTRAS for 'options'
         db_extras_dict = dict(parse_qsl(self.DB_EXTRAS))
         options = db_extras_dict.get("options", "")
-        # Always include timezone
-        timezone_opt = "-c timezone=UTC"
-        if options:
-            # Merge user options and timezone
-            merged_options = f"{options} {timezone_opt}"
-        else:
-            merged_options = timezone_opt
-
-        connect_args = {"options": merged_options}
+        connect_args = {}
+        # Use the dynamic SQLALCHEMY_DATABASE_URI_SCHEME property
+        if self.SQLALCHEMY_DATABASE_URI_SCHEME.startswith("postgresql"):
+            timezone_opt = "-c timezone=UTC"
+            if options:
+                merged_options = f"{options} {timezone_opt}"
+            else:
+                merged_options = timezone_opt
+            connect_args = {"options": merged_options}
 
         return {
             "pool_size": self.SQLALCHEMY_POOL_SIZE,

api/configs/middleware/vdb/oceanbase_config.py

@@ -7,29 +7,29 @@ class OceanBaseVectorConfig(BaseSettings):
     Configuration settings for OceanBase Vector database
     """
 
-    OCEANBASE_VECTOR_HOST: str | None = Field(
-        description="Hostname or IP address of the OceanBase Vector server (e.g. 'localhost')",
-        default=None,
+    OCEANBASE_HOST: str = Field(
+        description="OceanBase hostname or IP address.",
+        default="localhost",
     )
-
-    OCEANBASE_VECTOR_PORT: PositiveInt | None = Field(
-        description="Port number on which the OceanBase Vector server is listening (default is 2881)",
+    
+    OCEANBASE_PORT: PositiveInt = Field(
+        description="OceanBase port number.",
         default=2881,
     )
-
-    OCEANBASE_VECTOR_USER: str | None = Field(
-        description="Username for authenticating with the OceanBase Vector database",
-        default=None,
+    
+    OCEANBASE_USER: str = Field(
+        description="OceanBase username.",
+        default="root@test",
     )
-
-    OCEANBASE_VECTOR_PASSWORD: str | None = Field(
-        description="Password for authenticating with the OceanBase Vector database",
-        default=None,
+    
+    OCEANBASE_PASSWORD: str = Field(
+        description="OceanBase password.",
+        default="difyai123456",
     )
-
-    OCEANBASE_VECTOR_DATABASE: str | None = Field(
-        description="Name of the OceanBase Vector database to connect to",
-        default=None,
+    
+    OCEANBASE_DATABASE: str = Field(
+        description="OceanBase database name.",
+        default="test",
     )
 
     OCEANBASE_ENABLE_HYBRID_SEARCH: bool = Field(

api/controllers/console/__init__.py

@@ -58,13 +58,11 @@ from .app import (
     mcp_server,
     message,
     model_config,
-    online_user,
     ops_trace,
     site,
     statistic,
     workflow,
     workflow_app_log,
-    workflow_comment,
     workflow_draft_variable,
     workflow_run,
     workflow_statistic,
@@ -108,12 +106,10 @@ from .datasets.rag_pipeline import (
 
 # Import explore controllers
 from .explore import (
-    banner,
     installed_app,
     parameter,
     recommended_app,
     saved_message,
-    trial,
 )
 
 # Import tag controllers
@@ -147,7 +143,6 @@ __all__ = [
     "apikey",
     "app",
     "audio",
-    "banner",
     "billing",
     "bp",
     "completion",
@@ -201,7 +196,6 @@ __all__ = [
     "statistic",
     "tags",
     "tool_providers",
-    "trial",
     "version",
     "website",
     "workflow",

api/controllers/console/admin.py

@@ -16,7 +16,7 @@ from controllers.console import api, console_ns
 from controllers.console.wraps import only_edition_cloud
 from extensions.ext_database import db
 from libs.token import extract_access_token
-from models.model import App, ExporleBanner, InstalledApp, RecommendedApp, TrialApp
+from models.model import App, InstalledApp, RecommendedApp
 
 
 def admin_required(view: Callable[P, R]):
@@ -52,8 +52,6 @@ class InsertExploreAppListApi(Resource):
                 "language": fields.String(required=True, description="Language code"),
                 "category": fields.String(required=True, description="App category"),
                 "position": fields.Integer(required=True, description="Display position"),
-                "can_trial": fields.Boolean(required=True, description="Can trial"),
-                "trial_limit": fields.Integer(required=True, description="Trial limit"),
             },
         )
     )
@@ -73,8 +71,6 @@ class InsertExploreAppListApi(Resource):
             .add_argument("language", type=supported_language, required=True, nullable=False, location="json")
             .add_argument("category", type=str, required=True, nullable=False, location="json")
             .add_argument("position", type=int, required=True, nullable=False, location="json")
-            .add_argument("can_trial", type=bool, required=True, nullable=False, location="json")
-            .add_argument("trial_limit", type=int, required=True, nullable=False, location="json")
         )
         args = parser.parse_args()
 
@@ -112,20 +108,6 @@ class InsertExploreAppListApi(Resource):
                 )
 
                 db.session.add(recommended_app)
-                if args["can_trial"]:
-                    trial_app = db.session.execute(
-                        select(TrialApp).where(TrialApp.app_id == args["app_id"])
-                    ).scalar_one_or_none()
-                    if not trial_app:
-                        db.session.add(
-                            TrialApp(
-                                app_id=args["app_id"],
-                                tenant_id=app.tenant_id,
-                                trial_limit=args["trial_limit"],
-                            )
-                        )
-                    else:
-                        trial_app.trial_limit = args["trial_limit"]
 
                 app.is_public = True
                 db.session.commit()
@@ -140,20 +122,6 @@ class InsertExploreAppListApi(Resource):
                 recommended_app.category = args["category"]
                 recommended_app.position = args["position"]
 
-                if args["can_trial"]:
-                    trial_app = db.session.execute(
-                        select(TrialApp).where(TrialApp.app_id == args["app_id"])
-                    ).scalar_one_or_none()
-                    if not trial_app:
-                        db.session.add(
-                            TrialApp(
-                                app_id=args["app_id"],
-                                tenant_id=app.tenant_id,
-                                trial_limit=args["trial_limit"],
-                            )
-                        )
-                    else:
-                        trial_app.trial_limit = args["trial_limit"]
                 app.is_public = True
 
                 db.session.commit()
@@ -199,83 +167,7 @@ class InsertExploreAppApi(Resource):
             for installed_app in installed_apps:
                 session.delete(installed_app)
 
-            trial_app = session.execute(
-                select(TrialApp).where(TrialApp.app_id == recommended_app.app_id)
-            ).scalar_one_or_none()
-            if trial_app:
-                session.delete(trial_app)
-
         db.session.delete(recommended_app)
         db.session.commit()
 
         return {"result": "success"}, 204
-
-
-@console_ns.route("/admin/insert-explore-banner")
-class InsertExploreBanner(Resource):
-    @api.doc("insert_explore_banner")
-    @api.doc(description="Insert an explore banner")
-    @api.expect(
-        api.model(
-            "InsertExploreBannerRequest",
-            {
-                "content": fields.String(required=True, description="Banner content"),
-                "link": fields.String(required=True, description="Banner link"),
-                "sort": fields.Integer(required=True, description="Banner sort"),
-            },
-        )
-    )
-    @api.response(200, "Banner inserted successfully")
-    @admin_required
-    @only_edition_cloud
-    def post(self):
-        parser = reqparse.RequestParser()
-        parser.add_argument("category", type=str, required=True, nullable=False, location="json")
-        parser.add_argument("title", type=str, required=True, nullable=False, location="json")
-        parser.add_argument("description", type=str, required=True, nullable=False, location="json")
-        parser.add_argument("img-src", type=str, required=True, nullable=False, location="json")
-
-        parser.add_argument("language", type=str, required=True, nullable=False, location="json")
-        parser.add_argument("link", type=str, required=True, nullable=False, location="json")
-        parser.add_argument("sort", type=int, required=True, nullable=False, location="json")
-
-        args = parser.parse_args()
-
-        content = {
-            "category": args["category"],
-            "title": args["title"],
-            "description": args["description"],
-            "img-src": args["img-src"],
-        }
-
-        if not args["language"]:
-            args["language"] = "en-US"
-
-        banner = ExporleBanner(
-            content=content,
-            link=args["link"],
-            sort=args["sort"],
-            language=args["language"],
-        )
-        db.session.add(banner)
-        db.session.commit()
-
-        return {"result": "success"}, 200
-
-
-@console_ns.route("/admin/delete-explore-banner/<uuid:banner_id>")
-class DeleteExploreBanner(Resource):
-    @api.doc("delete_explore_banner")
-    @api.doc(description="Delete an explore banner")
-    @api.response(204, "Banner deleted successfully")
-    @admin_required
-    @only_edition_cloud
-    def delete(self, banner_id):
-        banner = db.session.execute(select(ExporleBanner).where(ExporleBanner.id == banner_id)).scalar_one_or_none()
-        if not banner:
-            raise NotFound(f"Banner '{banner_id}' is not found")
-
-        db.session.delete(banner)
-        db.session.commit()
-
-        return {"result": "success"}, 204

api/controllers/console/app/error.py

@@ -115,9 +115,3 @@ class InvokeRateLimitError(BaseHTTPException):
     error_code = "rate_limit_error"
     description = "Rate Limit Error"
     code = 429
-
-
-class NeedAddIdsError(BaseHTTPException):
-    error_code = "need_add_ids"
-    description = "Need to add ids."
-    code = 400

api/controllers/console/app/online_user.py

@@ -1,339 +0,0 @@
-import json
-import time
-
-from werkzeug.wrappers import Request as WerkzeugRequest
-
-from extensions.ext_redis import redis_client
-from extensions.ext_socketio import sio
-from libs.passport import PassportService
-from libs.token import extract_access_token
-from services.account_service import AccountService
-
-SESSION_STATE_TTL_SECONDS = 3600
-WORKFLOW_ONLINE_USERS_PREFIX = "workflow_online_users:"
-WORKFLOW_LEADER_PREFIX = "workflow_leader:"
-WS_SID_MAP_PREFIX = "ws_sid_map:"
-
-
-def _workflow_key(workflow_id: str) -> str:
-    return f"{WORKFLOW_ONLINE_USERS_PREFIX}{workflow_id}"
-
-
-def _leader_key(workflow_id: str) -> str:
-    return f"{WORKFLOW_LEADER_PREFIX}{workflow_id}"
-
-
-def _sid_key(sid: str) -> str:
-    return f"{WS_SID_MAP_PREFIX}{sid}"
-
-
-def _refresh_session_state(workflow_id: str, sid: str) -> None:
-    """
-    Refresh TTLs for workflow + session keys so healthy sessions do not linger forever after crashes.
-    """
-    workflow_key = _workflow_key(workflow_id)
-    sid_key = _sid_key(sid)
-    if redis_client.exists(workflow_key):
-        redis_client.expire(workflow_key, SESSION_STATE_TTL_SECONDS)
-    if redis_client.exists(sid_key):
-        redis_client.expire(sid_key, SESSION_STATE_TTL_SECONDS)
-
-
-@sio.on("connect")
-def socket_connect(sid, environ, auth):
-    """
-    WebSocket connect event, do authentication here.
-    """
-    token = None
-    if auth and isinstance(auth, dict):
-        token = auth.get("token")
-
-    if not token:
-        try:
-            request_environ = WerkzeugRequest(environ)
-            token = extract_access_token(request_environ)
-        except Exception:
-            token = None
-
-    if not token:
-        return False
-
-    try:
-        decoded = PassportService().verify(token)
-        user_id = decoded.get("user_id")
-        if not user_id:
-            return False
-
-        with sio.app.app_context():
-            user = AccountService.load_logged_in_account(account_id=user_id)
-            if not user:
-                return False
-
-            sio.save_session(sid, {"user_id": user.id, "username": user.name, "avatar": user.avatar})
-
-            return True
-
-    except Exception:
-        return False
-
-
-@sio.on("user_connect")
-def handle_user_connect(sid, data):
-    """
-    Handle user connect event. Each session (tab) is treated as an independent collaborator.
-    """
-
-    workflow_id = data.get("workflow_id")
-    if not workflow_id:
-        return {"msg": "workflow_id is required"}, 400
-
-    session = sio.get_session(sid)
-    user_id = session.get("user_id")
-
-    if not user_id:
-        return {"msg": "unauthorized"}, 401
-
-    # Each session is stored independently with sid as key
-    session_info = {
-        "user_id": user_id,
-        "username": session.get("username", "Unknown"),
-        "avatar": session.get("avatar", None),
-        "sid": sid,
-        "connected_at": int(time.time()),  # Add timestamp to differentiate tabs
-    }
-
-    workflow_key = _workflow_key(workflow_id)
-    # Store session info with sid as key
-    redis_client.hset(workflow_key, sid, json.dumps(session_info))
-    redis_client.set(
-        _sid_key(sid),
-        json.dumps({"workflow_id": workflow_id, "user_id": user_id}),
-        ex=SESSION_STATE_TTL_SECONDS,
-    )
-    _refresh_session_state(workflow_id, sid)
-
-    # Leader election: first session becomes the leader
-    leader_sid = get_or_set_leader(workflow_id, sid)
-    is_leader = leader_sid == sid
-
-    sio.enter_room(sid, workflow_id)
-    broadcast_online_users(workflow_id)
-
-    # Notify this session of their leader status
-    sio.emit("status", {"isLeader": is_leader}, room=sid)
-
-    return {"msg": "connected", "user_id": user_id, "sid": sid, "isLeader": is_leader}
-
-
-@sio.on("disconnect")
-def handle_disconnect(sid):
-    """
-    Handle session disconnect event. Remove the specific session from online users.
-    """
-    mapping = redis_client.get(_sid_key(sid))
-    if mapping:
-        data = json.loads(mapping)
-        workflow_id = data["workflow_id"]
-
-        # Remove this specific session
-        redis_client.hdel(_workflow_key(workflow_id), sid)
-        redis_client.delete(_sid_key(sid))
-
-        # Handle leader re-election if the leader session disconnected
-        handle_leader_disconnect(workflow_id, sid)
-
-        broadcast_online_users(workflow_id)
-
-
-def _clear_session_state(workflow_id: str, sid: str) -> None:
-    redis_client.hdel(_workflow_key(workflow_id), sid)
-    redis_client.delete(_sid_key(sid))
-
-
-def _is_session_active(workflow_id: str, sid: str) -> bool:
-    if not sid:
-        return False
-
-    try:
-        if not sio.manager.is_connected(sid, "/"):
-            return False
-    except AttributeError:
-        return False
-
-    if not redis_client.hexists(_workflow_key(workflow_id), sid):
-        return False
-
-    if not redis_client.exists(_sid_key(sid)):
-        return False
-
-    return True
-
-
-def get_or_set_leader(workflow_id: str, sid: str) -> str:
-    """
-    Get current leader session or set this session as leader if no valid leader exists.
-    Returns the leader session id (sid).
-    """
-    raw_leader = redis_client.get(_leader_key(workflow_id))
-    current_leader = raw_leader.decode("utf-8") if isinstance(raw_leader, bytes) else raw_leader
-    leader_replaced = False
-
-    if current_leader and not _is_session_active(workflow_id, current_leader):
-        _clear_session_state(workflow_id, current_leader)
-        redis_client.delete(_leader_key(workflow_id))
-        current_leader = None
-        leader_replaced = True
-
-    if not current_leader:
-        redis_client.set(_leader_key(workflow_id), sid, ex=SESSION_STATE_TTL_SECONDS)  # Expire in 1 hour
-        if leader_replaced:
-            broadcast_leader_change(workflow_id, sid)
-        return sid
-
-    return current_leader
-
-
-def handle_leader_disconnect(workflow_id, disconnected_sid):
-    """
-    Handle leader re-election when a session disconnects.
-    If the disconnected session was the leader, elect a new leader from remaining sessions.
-    """
-    current_leader = redis_client.get(_leader_key(workflow_id))
-
-    if current_leader:
-        current_leader = current_leader.decode("utf-8") if isinstance(current_leader, bytes) else current_leader
-
-        if current_leader == disconnected_sid:
-            # Leader session disconnected, elect a new leader
-            sessions_json = redis_client.hgetall(_workflow_key(workflow_id))
-
-            if sessions_json:
-                # Get the first remaining session as new leader
-                new_leader_sid = list(sessions_json.keys())[0]
-                if isinstance(new_leader_sid, bytes):
-                    new_leader_sid = new_leader_sid.decode("utf-8")
-
-                redis_client.set(_leader_key(workflow_id), new_leader_sid, ex=SESSION_STATE_TTL_SECONDS)
-
-                # Notify all sessions about the new leader
-                broadcast_leader_change(workflow_id, new_leader_sid)
-            else:
-                # No sessions left, remove leader
-                redis_client.delete(_leader_key(workflow_id))
-
-
-def broadcast_leader_change(workflow_id, new_leader_sid):
-    """
-    Broadcast leader change to all sessions in the workflow.
-    """
-    sessions_json = redis_client.hgetall(_workflow_key(workflow_id))
-
-    for sid, session_info_json in sessions_json.items():
-        try:
-            sid_str = sid.decode("utf-8") if isinstance(sid, bytes) else sid
-            is_leader = sid_str == new_leader_sid
-            # Emit to each session whether they are the new leader
-            sio.emit("status", {"isLeader": is_leader}, room=sid_str)
-        except Exception:
-            continue
-
-
-def get_current_leader(workflow_id):
-    """
-    Get the current leader for a workflow.
-    """
-    leader = redis_client.get(_leader_key(workflow_id))
-    return leader.decode("utf-8") if leader and isinstance(leader, bytes) else leader
-
-
-def broadcast_online_users(workflow_id):
-    """
-    Broadcast online users to the workflow room.
-    Each session is shown as a separate user (even if same person has multiple tabs).
-    """
-    sessions_json = redis_client.hgetall(_workflow_key(workflow_id))
-    users = []
-
-    for sid, session_info_json in sessions_json.items():
-        try:
-            session_info = json.loads(session_info_json)
-            # Each session appears as a separate "user" in the UI
-            users.append(
-                {
-                    "user_id": session_info["user_id"],
-                    "username": session_info["username"],
-                    "avatar": session_info.get("avatar"),
-                    "sid": session_info["sid"],
-                    "connected_at": session_info.get("connected_at"),
-                }
-            )
-        except Exception:
-            continue
-
-    # Sort by connection time to maintain consistent order
-    users.sort(key=lambda x: x.get("connected_at") or 0)
-
-    # Get current leader session
-    leader_sid = get_current_leader(workflow_id)
-
-    sio.emit("online_users", {"workflow_id": workflow_id, "users": users, "leader": leader_sid}, room=workflow_id)
-
-
-@sio.on("collaboration_event")
-def handle_collaboration_event(sid, data):
-    """
-    Handle general collaboration events, include:
-    1. mouse_move
-    2. vars_and_features_update
-    3. sync_request (ask leader to update graph)
-    4. app_state_update
-    5. mcp_server_update
-    6. workflow_update
-    7. comments_update
-    8. node_panel_presence
-
-    """
-    mapping = redis_client.get(_sid_key(sid))
-
-    if not mapping:
-        return {"msg": "unauthorized"}, 401
-
-    mapping_data = json.loads(mapping)
-    workflow_id = mapping_data["workflow_id"]
-    user_id = mapping_data["user_id"]
-    _refresh_session_state(workflow_id, sid)
-
-    event_type = data.get("type")
-    event_data = data.get("data")
-    timestamp = data.get("timestamp", int(time.time()))
-
-    if not event_type:
-        return {"msg": "invalid event type"}, 400
-
-    sio.emit(
-        "collaboration_update",
-        {"type": event_type, "userId": user_id, "data": event_data, "timestamp": timestamp},
-        room=workflow_id,
-        skip_sid=sid,
-    )
-
-    return {"msg": "event_broadcasted"}
-
-
-@sio.on("graph_event")
-def handle_graph_event(sid, data):
-    """
-    Handle graph events - simple broadcast relay.
-    """
-    mapping = redis_client.get(_sid_key(sid))
-
-    if not mapping:
-        return {"msg": "unauthorized"}, 401
-
-    mapping_data = json.loads(mapping)
-    workflow_id = mapping_data["workflow_id"]
-    _refresh_session_state(workflow_id, sid)
-
-    sio.emit("graph_update", data, room=workflow_id, skip_sid=sid)
-
-    return {"msg": "graph_update_broadcasted"}

api/controllers/console/app/statistic.py

@@ -10,9 +10,9 @@ from controllers.console.wraps import account_initialization_required, setup_req
 from core.app.entities.app_invoke_entities import InvokeFrom
 from extensions.ext_database import db
 from libs.datetime_utils import parse_time_range
-from libs.helper import DatetimeString
+from libs.helper import DatetimeString, convert_datetime_to_date
 from libs.login import current_account_with_tenant, login_required
-from models import AppMode, Message
+from models import AppMode
 
 
 @console_ns.route("/apps/<uuid:app_id>/statistics/daily-messages")
@@ -44,8 +44,9 @@ class DailyMessageStatistic(Resource):
         )
         args = parser.parse_args()
 
-        sql_query = """SELECT
-    DATE(DATE_TRUNC('day', created_at AT TIME ZONE 'UTC' AT TIME ZONE :tz )) AS date,
+        converted_created_at = convert_datetime_to_date("created_at")
+        sql_query = f"""SELECT
+    {converted_created_at} AS date,
     COUNT(*) AS message_count
 FROM
     messages
@@ -108,6 +109,17 @@ class DailyConversationStatistic(Resource):
             .add_argument("end", type=DatetimeString("%Y-%m-%d %H:%M"), location="args")
         )
         args = parser.parse_args()
+
+        converted_created_at = convert_datetime_to_date("created_at")
+        sql_query = f"""SELECT
+    {converted_created_at} AS date,
+    COUNT(DISTINCT conversation_id) AS conversation_count
+FROM
+    messages
+WHERE
+    app_id = :app_id
+    AND invoke_from != :invoke_from"""
+        arg_dict = {"tz": account.timezone, "app_id": app_model.id, "invoke_from": InvokeFrom.DEBUGGER}
         assert account.timezone is not None
 
         try:
@@ -115,30 +127,21 @@ class DailyConversationStatistic(Resource):
         except ValueError as e:
             abort(400, description=str(e))
 
-        stmt = (
-            sa.select(
-                sa.func.date(
-                    sa.func.date_trunc("day", sa.text("created_at AT TIME ZONE 'UTC' AT TIME ZONE :tz"))
-                ).label("date"),
-                sa.func.count(sa.distinct(Message.conversation_id)).label("conversation_count"),
-            )
-            .select_from(Message)
-            .where(Message.app_id == app_model.id, Message.invoke_from != InvokeFrom.DEBUGGER)
-        )
-
         if start_datetime_utc:
-            stmt = stmt.where(Message.created_at >= start_datetime_utc)
+            sql_query += " AND created_at >= :start"
+            arg_dict["start"] = start_datetime_utc
 
         if end_datetime_utc:
-            stmt = stmt.where(Message.created_at < end_datetime_utc)
+            sql_query += " AND created_at < :end"
+            arg_dict["end"] = end_datetime_utc
 
-        stmt = stmt.group_by("date").order_by("date")
+        sql_query += " GROUP BY date ORDER BY date"
 
         response_data = []
         with db.engine.begin() as conn:
-            rs = conn.execute(stmt, {"tz": account.timezone})
-            for row in rs:
-                response_data.append({"date": str(row.date), "conversation_count": row.conversation_count})
+            rs = conn.execute(sa.text(sql_query), arg_dict)
+            for i in rs:
+                response_data.append({"date": str(i.date), "conversation_count": i.conversation_count})
 
         return jsonify({"data": response_data})
 
@@ -172,8 +175,9 @@ class DailyTerminalsStatistic(Resource):
         )
         args = parser.parse_args()
 
-        sql_query = """SELECT
-    DATE(DATE_TRUNC('day', created_at AT TIME ZONE 'UTC' AT TIME ZONE :tz )) AS date,
+        converted_created_at = convert_datetime_to_date("created_at")
+        sql_query = f"""SELECT
+    {converted_created_at} AS date,
     COUNT(DISTINCT messages.from_end_user_id) AS terminal_count
 FROM
     messages
@@ -237,8 +241,9 @@ class DailyTokenCostStatistic(Resource):
         )
         args = parser.parse_args()
 
-        sql_query = """SELECT
-    DATE(DATE_TRUNC('day', created_at AT TIME ZONE 'UTC' AT TIME ZONE :tz )) AS date,
+        converted_created_at = convert_datetime_to_date("created_at")
+        sql_query = f"""SELECT
+    {converted_created_at} AS date,
     (SUM(messages.message_tokens) + SUM(messages.answer_tokens)) AS token_count,
     SUM(total_price) AS total_price
 FROM
@@ -305,8 +310,9 @@ class AverageSessionInteractionStatistic(Resource):
         )
         args = parser.parse_args()
 
-        sql_query = """SELECT
-    DATE(DATE_TRUNC('day', c.created_at AT TIME ZONE 'UTC' AT TIME ZONE :tz )) AS date,
+        converted_created_at = convert_datetime_to_date("c.created_at")
+        sql_query = f"""SELECT
+    {converted_created_at} AS date,
     AVG(subquery.message_count) AS interactions
 FROM
     (
@@ -389,8 +395,9 @@ class UserSatisfactionRateStatistic(Resource):
         )
         args = parser.parse_args()
 
-        sql_query = """SELECT
-    DATE(DATE_TRUNC('day', m.created_at AT TIME ZONE 'UTC' AT TIME ZONE :tz )) AS date,
+        converted_created_at = convert_datetime_to_date("m.created_at")
+        sql_query = f"""SELECT
+    {converted_created_at} AS date,
     COUNT(m.id) AS message_count,
     COUNT(mf.id) AS feedback_count
 FROM
@@ -463,8 +470,9 @@ class AverageResponseTimeStatistic(Resource):
         )
         args = parser.parse_args()
 
-        sql_query = """SELECT
-    DATE(DATE_TRUNC('day', created_at AT TIME ZONE 'UTC' AT TIME ZONE :tz )) AS date,
+        converted_created_at = convert_datetime_to_date("created_at")
+        sql_query = f"""SELECT
+    {converted_created_at} AS date,
     AVG(provider_response_latency) AS latency
 FROM
     messages
@@ -528,8 +536,9 @@ class TokensPerSecondStatistic(Resource):
         )
         args = parser.parse_args()
 
-        sql_query = """SELECT
-    DATE(DATE_TRUNC('day', created_at AT TIME ZONE 'UTC' AT TIME ZONE :tz )) AS date,
+        converted_created_at = convert_datetime_to_date("created_at")
+        sql_query = f"""SELECT
+    {converted_created_at} AS date,
     CASE
         WHEN SUM(provider_response_latency) = 0 THEN 0
         ELSE (SUM(answer_tokens) / SUM(provider_response_latency))

api/controllers/console/app/workflow.py

@@ -5,12 +5,10 @@ from typing import cast
 
 from flask import abort, request
 from flask_restx import Resource, fields, inputs, marshal_with, reqparse
-from pydantic_core import ValidationError
 from sqlalchemy.orm import Session
 from werkzeug.exceptions import Forbidden, InternalServerError, NotFound
 
 import services
-from configs import dify_config
 from controllers.console import api, console_ns
 from controllers.console.app.error import ConversationCompletedError, DraftWorkflowNotExist, DraftWorkflowNotSync
 from controllers.console.app.wraps import get_app_model
@@ -23,9 +21,7 @@ from core.file.models import File
 from core.helper.trace_id_helper import get_external_trace_id
 from core.workflow.graph_engine.manager import GraphEngineManager
 from extensions.ext_database import db
-from extensions.ext_redis import redis_client
 from factories import file_factory, variable_factory
-from fields.online_user_fields import online_user_list_fields
 from fields.workflow_fields import workflow_fields, workflow_pagination_fields
 from fields.workflow_run_fields import workflow_run_node_execution_fields
 from libs import helper
@@ -103,7 +99,6 @@ class DraftWorkflowApi(Resource):
                 "hash": fields.String(description="Workflow hash for validation"),
                 "environment_variables": fields.List(fields.Raw, required=True, description="Environment variables"),
                 "conversation_variables": fields.List(fields.Raw, description="Conversation variables"),
-                "memory_blocks": fields.List(fields.Raw, description="Memory blocks"),
             },
         )
     )
@@ -138,8 +133,6 @@ class DraftWorkflowApi(Resource):
                 .add_argument("hash", type=str, required=False, location="json")
                 .add_argument("environment_variables", type=list, required=True, location="json")
                 .add_argument("conversation_variables", type=list, required=False, location="json")
-                .add_argument("force_upload", type=bool, required=False, default=False, location="json")
-                .add_argument("memory_blocks", type=list, required=False, location="json")
             )
             args = parser.parse_args()
         elif "text/plain" in content_type:
@@ -157,8 +150,6 @@ class DraftWorkflowApi(Resource):
                     "hash": data.get("hash"),
                     "environment_variables": data.get("environment_variables"),
                     "conversation_variables": data.get("conversation_variables"),
-                    "memory_blocks": data.get("memory_blocks"),
-                    "force_upload": data.get("force_upload", False),
                 }
             except json.JSONDecodeError:
                 return {"message": "Invalid JSON data"}, 400
@@ -175,10 +166,6 @@ class DraftWorkflowApi(Resource):
             conversation_variables = [
                 variable_factory.build_conversation_variable_from_mapping(obj) for obj in conversation_variables_list
             ]
-            memory_blocks_list = args.get("memory_blocks") or []
-            from core.memory.entities import MemoryBlockSpec
-
-            memory_blocks = [MemoryBlockSpec.model_validate(obj) for obj in memory_blocks_list]
             workflow = workflow_service.sync_draft_workflow(
                 app_model=app_model,
                 graph=args["graph"],
@@ -187,13 +174,9 @@ class DraftWorkflowApi(Resource):
                 account=current_user,
                 environment_variables=environment_variables,
                 conversation_variables=conversation_variables,
-                force_upload=args.get("force_upload", False),
-                memory_blocks=memory_blocks,
             )
         except WorkflowHashNotEqualError:
             raise DraftWorkflowNotSync()
-        except ValidationError as e:
-            return {"message": str(e)}, 400
 
         return {
             "result": "success",
@@ -762,45 +745,6 @@ class ConvertToWorkflowApi(Resource):
         }
 
 
-@console_ns.route("/apps/<uuid:app_id>/workflows/draft/config")
-class WorkflowConfigApi(Resource):
-    """Resource for workflow configuration."""
-
-    @api.doc("get_workflow_config")
-    @api.doc(description="Get workflow configuration")
-    @api.doc(params={"app_id": "Application ID"})
-    @api.response(200, "Workflow configuration retrieved successfully")
-    @setup_required
-    @login_required
-    @account_initialization_required
-    @get_app_model(mode=[AppMode.ADVANCED_CHAT, AppMode.WORKFLOW])
-    def get(self, app_model: App):
-        return {
-            "parallel_depth_limit": dify_config.WORKFLOW_PARALLEL_DEPTH_LIMIT,
-        }
-
-
-class WorkflowFeaturesApi(Resource):
-    """Update draft workflow features."""
-
-    @setup_required
-    @login_required
-    @account_initialization_required
-    @get_app_model(mode=[AppMode.ADVANCED_CHAT, AppMode.WORKFLOW])
-    def post(self, app_model: App):
-        parser = reqparse.RequestParser()
-        parser.add_argument("features", type=dict, required=True, location="json")
-        args = parser.parse_args()
-
-        features = args.get("features")
-
-        # Update draft workflow features
-        workflow_service = WorkflowService()
-        workflow_service.update_draft_workflow_features(app_model=app_model, features=features, account=current_user)
-
-        return {"result": "success"}
-
-
 @console_ns.route("/apps/<uuid:app_id>/workflows")
 class PublishedAllWorkflowApi(Resource):
     @api.doc("get_all_published_workflows")
@@ -982,105 +926,3 @@ class DraftWorkflowNodeLastRunApi(Resource):
         if node_exec is None:
             raise NotFound("last run not found")
         return node_exec
-
-
-class WorkflowOnlineUsersApi(Resource):
-    @setup_required
-    @login_required
-    @account_initialization_required
-    @marshal_with(online_user_list_fields)
-    def get(self):
-        parser = reqparse.RequestParser()
-        parser.add_argument("workflow_ids", type=str, required=True, location="args")
-        args = parser.parse_args()
-
-        workflow_ids = [id.strip() for id in args["workflow_ids"].split(",")]
-
-        results = []
-        for workflow_id in workflow_ids:
-            users_json = redis_client.hgetall(f"workflow_online_users:{workflow_id}")
-
-            users = []
-            for _, user_info_json in users_json.items():
-                try:
-                    users.append(json.loads(user_info_json))
-                except Exception:
-                    continue
-            results.append({"workflow_id": workflow_id, "users": users})
-
-        return {"data": results}
-
-
-api.add_resource(
-    DraftWorkflowApi,
-    "/apps/<uuid:app_id>/workflows/draft",
-)
-api.add_resource(
-    WorkflowConfigApi,
-    "/apps/<uuid:app_id>/workflows/draft/config",
-)
-api.add_resource(
-    WorkflowFeaturesApi,
-    "/apps/<uuid:app_id>/workflows/draft/features",
-)
-api.add_resource(
-    AdvancedChatDraftWorkflowRunApi,
-    "/apps/<uuid:app_id>/advanced-chat/workflows/draft/run",
-)
-api.add_resource(
-    DraftWorkflowRunApi,
-    "/apps/<uuid:app_id>/workflows/draft/run",
-)
-api.add_resource(
-    WorkflowTaskStopApi,
-    "/apps/<uuid:app_id>/workflow-runs/tasks/<string:task_id>/stop",
-)
-api.add_resource(
-    DraftWorkflowNodeRunApi,
-    "/apps/<uuid:app_id>/workflows/draft/nodes/<string:node_id>/run",
-)
-api.add_resource(
-    AdvancedChatDraftRunIterationNodeApi,
-    "/apps/<uuid:app_id>/advanced-chat/workflows/draft/iteration/nodes/<string:node_id>/run",
-)
-api.add_resource(
-    WorkflowDraftRunIterationNodeApi,
-    "/apps/<uuid:app_id>/workflows/draft/iteration/nodes/<string:node_id>/run",
-)
-api.add_resource(
-    AdvancedChatDraftRunLoopNodeApi,
-    "/apps/<uuid:app_id>/advanced-chat/workflows/draft/loop/nodes/<string:node_id>/run",
-)
-api.add_resource(
-    WorkflowDraftRunLoopNodeApi,
-    "/apps/<uuid:app_id>/workflows/draft/loop/nodes/<string:node_id>/run",
-)
-api.add_resource(
-    PublishedWorkflowApi,
-    "/apps/<uuid:app_id>/workflows/publish",
-)
-api.add_resource(
-    PublishedAllWorkflowApi,
-    "/apps/<uuid:app_id>/workflows",
-)
-api.add_resource(
-    DefaultBlockConfigsApi,
-    "/apps/<uuid:app_id>/workflows/default-workflow-block-configs",
-)
-api.add_resource(
-    DefaultBlockConfigApi,
-    "/apps/<uuid:app_id>/workflows/default-workflow-block-configs/<string:block_type>",
-)
-api.add_resource(
-    ConvertToWorkflowApi,
-    "/apps/<uuid:app_id>/convert-to-workflow",
-)
-api.add_resource(
-    WorkflowByIdApi,
-    "/apps/<uuid:app_id>/workflows/<string:workflow_id>",
-)
-api.add_resource(
-    DraftWorkflowNodeLastRunApi,
-    "/apps/<uuid:app_id>/workflows/draft/nodes/<string:node_id>/last-run",
-)
-api.add_resource(WorkflowOnlineUsersApi, "/apps/workflows/online-users")

api/controllers/console/app/workflow_comment.py

@@ -1,240 +0,0 @@
-import logging
-
-from flask_restx import Resource, fields, marshal_with, reqparse
-
-from controllers.console import api
-from controllers.console.app.wraps import get_app_model
-from controllers.console.wraps import account_initialization_required, setup_required
-from fields.member_fields import account_with_role_fields
-from fields.workflow_comment_fields import (
-    workflow_comment_basic_fields,
-    workflow_comment_create_fields,
-    workflow_comment_detail_fields,
-    workflow_comment_reply_create_fields,
-    workflow_comment_reply_update_fields,
-    workflow_comment_resolve_fields,
-    workflow_comment_update_fields,
-)
-from libs.login import current_user, login_required
-from models import App
-from services.account_service import TenantService
-from services.workflow_comment_service import WorkflowCommentService
-
-logger = logging.getLogger(__name__)
-
-
-class WorkflowCommentListApi(Resource):
-    """API for listing and creating workflow comments."""
-
-    @login_required
-    @setup_required
-    @account_initialization_required
-    @get_app_model
-    @marshal_with(workflow_comment_basic_fields, envelope="data")
-    def get(self, app_model: App):
-        """Get all comments for a workflow."""
-        comments = WorkflowCommentService.get_comments(tenant_id=current_user.current_tenant_id, app_id=app_model.id)
-
-        return comments
-
-    @login_required
-    @setup_required
-    @account_initialization_required
-    @get_app_model
-    @marshal_with(workflow_comment_create_fields)
-    def post(self, app_model: App):
-        """Create a new workflow comment."""
-        parser = reqparse.RequestParser()
-        parser.add_argument("position_x", type=float, required=True, location="json")
-        parser.add_argument("position_y", type=float, required=True, location="json")
-        parser.add_argument("content", type=str, required=True, location="json")
-        parser.add_argument("mentioned_user_ids", type=list, location="json", default=[])
-        args = parser.parse_args()
-
-        result = WorkflowCommentService.create_comment(
-            tenant_id=current_user.current_tenant_id,
-            app_id=app_model.id,
-            created_by=current_user.id,
-            content=args.content,
-            position_x=args.position_x,
-            position_y=args.position_y,
-            mentioned_user_ids=args.mentioned_user_ids,
-        )
-
-        return result, 201
-
-
-class WorkflowCommentDetailApi(Resource):
-    """API for managing individual workflow comments."""
-
-    @login_required
-    @setup_required
-    @account_initialization_required
-    @get_app_model
-    @marshal_with(workflow_comment_detail_fields)
-    def get(self, app_model: App, comment_id: str):
-        """Get a specific workflow comment."""
-        comment = WorkflowCommentService.get_comment(
-            tenant_id=current_user.current_tenant_id, app_id=app_model.id, comment_id=comment_id
-        )
-
-        return comment
-
-    @login_required
-    @setup_required
-    @account_initialization_required
-    @get_app_model
-    @marshal_with(workflow_comment_update_fields)
-    def put(self, app_model: App, comment_id: str):
-        """Update a workflow comment."""
-        parser = reqparse.RequestParser()
-        parser.add_argument("content", type=str, required=True, location="json")
-        parser.add_argument("position_x", type=float, required=False, location="json")
-        parser.add_argument("position_y", type=float, required=False, location="json")
-        parser.add_argument("mentioned_user_ids", type=list, location="json", default=[])
-        args = parser.parse_args()
-
-        result = WorkflowCommentService.update_comment(
-            tenant_id=current_user.current_tenant_id,
-            app_id=app_model.id,
-            comment_id=comment_id,
-            user_id=current_user.id,
-            content=args.content,
-            position_x=args.position_x,
-            position_y=args.position_y,
-            mentioned_user_ids=args.mentioned_user_ids,
-        )
-
-        return result
-
-    @login_required
-    @setup_required
-    @account_initialization_required
-    @get_app_model
-    def delete(self, app_model: App, comment_id: str):
-        """Delete a workflow comment."""
-        WorkflowCommentService.delete_comment(
-            tenant_id=current_user.current_tenant_id,
-            app_id=app_model.id,
-            comment_id=comment_id,
-            user_id=current_user.id,
-        )
-
-        return {"result": "success"}, 204
-
-
-class WorkflowCommentResolveApi(Resource):
-    """API for resolving and reopening workflow comments."""
-
-    @login_required
-    @setup_required
-    @account_initialization_required
-    @get_app_model
-    @marshal_with(workflow_comment_resolve_fields)
-    def post(self, app_model: App, comment_id: str):
-        """Resolve a workflow comment."""
-        comment = WorkflowCommentService.resolve_comment(
-            tenant_id=current_user.current_tenant_id,
-            app_id=app_model.id,
-            comment_id=comment_id,
-            user_id=current_user.id,
-        )
-
-        return comment
-
-
-class WorkflowCommentReplyApi(Resource):
-    """API for managing comment replies."""
-
-    @login_required
-    @setup_required
-    @account_initialization_required
-    @get_app_model
-    @marshal_with(workflow_comment_reply_create_fields)
-    def post(self, app_model: App, comment_id: str):
-        """Add a reply to a workflow comment."""
-        # Validate comment access first
-        WorkflowCommentService.validate_comment_access(
-            comment_id=comment_id, tenant_id=current_user.current_tenant_id, app_id=app_model.id
-        )
-
-        parser = reqparse.RequestParser()
-        parser.add_argument("content", type=str, required=True, location="json")
-        parser.add_argument("mentioned_user_ids", type=list, location="json", default=[])
-        args = parser.parse_args()
-
-        result = WorkflowCommentService.create_reply(
-            comment_id=comment_id,
-            content=args.content,
-            created_by=current_user.id,
-            mentioned_user_ids=args.mentioned_user_ids,
-        )
-
-        return result, 201
-
-
-class WorkflowCommentReplyDetailApi(Resource):
-    """API for managing individual comment replies."""
-
-    @login_required
-    @setup_required
-    @account_initialization_required
-    @get_app_model
-    @marshal_with(workflow_comment_reply_update_fields)
-    def put(self, app_model: App, comment_id: str, reply_id: str):
-        """Update a comment reply."""
-        # Validate comment access first
-        WorkflowCommentService.validate_comment_access(
-            comment_id=comment_id, tenant_id=current_user.current_tenant_id, app_id=app_model.id
-        )
-
-        parser = reqparse.RequestParser()
-        parser.add_argument("content", type=str, required=True, location="json")
-        parser.add_argument("mentioned_user_ids", type=list, location="json", default=[])
-        args = parser.parse_args()
-
-        reply = WorkflowCommentService.update_reply(
-            reply_id=reply_id, user_id=current_user.id, content=args.content, mentioned_user_ids=args.mentioned_user_ids
-        )
-
-        return reply
-
-    @login_required
-    @setup_required
-    @account_initialization_required
-    @get_app_model
-    def delete(self, app_model: App, comment_id: str, reply_id: str):
-        """Delete a comment reply."""
-        # Validate comment access first
-        WorkflowCommentService.validate_comment_access(
-            comment_id=comment_id, tenant_id=current_user.current_tenant_id, app_id=app_model.id
-        )
-
-        WorkflowCommentService.delete_reply(reply_id=reply_id, user_id=current_user.id)
-
-        return {"result": "success"}, 204
-
-
-class WorkflowCommentMentionUsersApi(Resource):
-    """API for getting mentionable users for workflow comments."""
-
-    @login_required
-    @setup_required
-    @account_initialization_required
-    @get_app_model
-    @marshal_with({"users": fields.List(fields.Nested(account_with_role_fields))})
-    def get(self, app_model: App):
-        """Get all users in current tenant for mentions."""
-        members = TenantService.get_tenant_members(current_user.current_tenant)
-        return {"users": members}
-
-
-# Register API routes
-api.add_resource(WorkflowCommentListApi, "/apps/<uuid:app_id>/workflow/comments")
-api.add_resource(WorkflowCommentDetailApi, "/apps/<uuid:app_id>/workflow/comments/<string:comment_id>")
-api.add_resource(WorkflowCommentResolveApi, "/apps/<uuid:app_id>/workflow/comments/<string:comment_id>/resolve")
-api.add_resource(WorkflowCommentReplyApi, "/apps/<uuid:app_id>/workflow/comments/<string:comment_id>/replies")
-api.add_resource(
-    WorkflowCommentReplyDetailApi, "/apps/<uuid:app_id>/workflow/comments/<string:comment_id>/replies/<string:reply_id>"
-)
-api.add_resource(WorkflowCommentMentionUsersApi, "/apps/<uuid:app_id>/workflow/comments/mention-users")

api/controllers/console/app/workflow_draft_variable.py

@@ -19,8 +19,8 @@ from core.variables.segments import ArrayFileSegment, FileSegment, Segment
 from core.variables.types import SegmentType
 from core.workflow.constants import CONVERSATION_VARIABLE_NODE_ID, SYSTEM_VARIABLE_NODE_ID
 from extensions.ext_database import db
-from factories import variable_factory
 from factories.file_factory import build_from_mapping, build_from_mappings
+from factories.variable_factory import build_segment_with_type
 from libs.login import current_user, login_required
 from models import Account, App, AppMode
 from models.workflow import WorkflowDraftVariable
@@ -355,7 +355,7 @@ class VariableApi(Resource):
                 if len(raw_value) > 0 and not isinstance(raw_value[0], dict):
                     raise InvalidArgumentError(description=f"expected dict for files[0], got {type(raw_value)}")
                 raw_value = build_from_mappings(mappings=raw_value, tenant_id=app_model.tenant_id)
-            new_value = variable_factory.build_segment_with_type(variable.value_type, raw_value)
+            new_value = build_segment_with_type(variable.value_type, raw_value)
         draft_var_srv.update_variable(variable, name=new_name, value=new_value)
         db.session.commit()
         return variable
@@ -448,35 +448,8 @@ class ConversationVariableCollectionApi(Resource):
         db.session.commit()
         return _get_variable_list(app_model, CONVERSATION_VARIABLE_NODE_ID)
 
-    @setup_required
-    @login_required
-    @account_initialization_required
-    @get_app_model(mode=AppMode.ADVANCED_CHAT)
-    def post(self, app_model: App):
-        # The role of the current user in the ta table must be admin, owner, or editor
-        if not current_user.is_editor:
-            raise Forbidden()
-
-        parser = reqparse.RequestParser()
-        parser.add_argument("conversation_variables", type=list, required=True, location="json")
-        args = parser.parse_args()
-
-        workflow_service = WorkflowService()
-
-        conversation_variables_list = args.get("conversation_variables") or []
-        conversation_variables = [
-            variable_factory.build_conversation_variable_from_mapping(obj) for obj in conversation_variables_list
-        ]
-
-        workflow_service.update_draft_workflow_conversation_variables(
-            app_model=app_model,
-            account=current_user,
-            conversation_variables=conversation_variables,
-        )
-
-        return {"result": "success"}
-
 
+@console_ns.route("/apps/<uuid:app_id>/workflows/draft/system-variables")
 class SystemVariableCollectionApi(Resource):
     @api.doc("get_system_variables")
     @api.doc(description="Get system variables for workflow")
@@ -526,44 +499,3 @@ class EnvironmentVariableCollectionApi(Resource):
             )
 
         return {"items": env_vars_list}
-
-    @setup_required
-    @login_required
-    @account_initialization_required
-    @get_app_model(mode=[AppMode.ADVANCED_CHAT, AppMode.WORKFLOW])
-    def post(self, app_model: App):
-        # The role of the current user in the ta table must be admin, owner, or editor
-        if not current_user.is_editor:
-            raise Forbidden()
-
-        parser = reqparse.RequestParser()
-        parser.add_argument("environment_variables", type=list, required=True, location="json")
-        args = parser.parse_args()
-
-        workflow_service = WorkflowService()
-
-        environment_variables_list = args.get("environment_variables") or []
-        environment_variables = [
-            variable_factory.build_environment_variable_from_mapping(obj) for obj in environment_variables_list
-        ]
-
-        workflow_service.update_draft_workflow_environment_variables(
-            app_model=app_model,
-            account=current_user,
-            environment_variables=environment_variables,
-        )
-
-        return {"result": "success"}
-
-
-api.add_resource(
-    WorkflowVariableCollectionApi,
-    "/apps/<uuid:app_id>/workflows/draft/variables",
-)
-api.add_resource(NodeVariableCollectionApi, "/apps/<uuid:app_id>/workflows/draft/nodes/<string:node_id>/variables")
-api.add_resource(VariableApi, "/apps/<uuid:app_id>/workflows/draft/variables/<uuid:variable_id>")
-api.add_resource(VariableResetApi, "/apps/<uuid:app_id>/workflows/draft/variables/<uuid:variable_id>/reset")
-
-api.add_resource(ConversationVariableCollectionApi, "/apps/<uuid:app_id>/workflows/draft/conversation-variables")
-api.add_resource(SystemVariableCollectionApi, "/apps/<uuid:app_id>/workflows/draft/system-variables")
-api.add_resource(EnvironmentVariableCollectionApi, "/apps/<uuid:app_id>/workflows/draft/environment-variables")

api/controllers/console/app/wraps.py

@@ -23,15 +23,6 @@ def _load_app_model(app_id: str) -> App | None:
     return app_model
 
 
-def _load_app_model_with_trial(app_id: str) -> App | None:
-    app_model = (
-        db.session.query(App)
-        .where(App.id == app_id, App.status == "normal")
-        .first()
-    )
-    return app_model
-
-
 def get_app_model(view: Callable[P, R] | None = None, *, mode: Union[AppMode, list[AppMode], None] = None):
     def decorator(view_func: Callable[P1, R1]):
         @wraps(view_func)
@@ -71,44 +62,3 @@ def get_app_model(view: Callable[P, R] | None = None, *, mode: Union[AppMode, li
         return decorator
     else:
         return decorator(view)
-
-
-def get_app_model_with_trial(view: Callable[P, R] | None = None, *, mode: Union[AppMode, list[AppMode], None] = None):
-    def decorator(view_func: Callable[P, R]):
-        @wraps(view_func)
-        def decorated_view(*args: P.args, **kwargs: P.kwargs):
-            if not kwargs.get("app_id"):
-                raise ValueError("missing app_id in path parameters")
-
-            app_id = kwargs.get("app_id")
-            app_id = str(app_id)
-
-            del kwargs["app_id"]
-
-            app_model = _load_app_model_with_trial(app_id)
-
-            if not app_model:
-                raise AppNotFoundError()
-
-            app_mode = AppMode.value_of(app_model.mode)
-
-            if mode is not None:
-                if isinstance(mode, list):
-                    modes = mode
-                else:
-                    modes = [mode]
-
-                if app_mode not in modes:
-                    mode_values = {m.value for m in modes}
-                    raise AppNotFoundError(f"App mode is not in the supported list: {mode_values}")
-
-            kwargs["app_model"] = app_model
-
-            return view_func(*args, **kwargs)
-
-        return decorated_view
-
-    if view is None:
-        return decorator
-    else:
-        return decorator(view)

api/controllers/console/auth/login.py

@@ -25,13 +25,10 @@ from controllers.console.wraps import email_password_login_enabled, setup_requir
 from events.tenant_event import tenant_was_created
 from libs.helper import email, extract_remote_ip
 from libs.login import current_account_with_tenant
-from libs.passport import PassportService
 from libs.token import (
-    check_csrf_token,
     clear_access_token_from_cookie,
     clear_csrf_token_from_cookie,
     clear_refresh_token_from_cookie,
-    extract_access_token,
     extract_refresh_token,
     set_access_token_to_cookie,
     set_csrf_token_to_cookie,
@@ -292,18 +289,3 @@ class RefreshTokenApi(Resource):
             return response
         except Exception as e:
             return {"result": "fail", "message": str(e)}, 401
-
-
-# this api helps frontend to check whether user is authenticated
-# TODO: remove in the future. frontend should redirect to login page by catching 401 status
-@console_ns.route("/login/status")
-class LoginStatus(Resource):
-    def get(self):
-        token = extract_access_token(request)
-        res = True
-        try:
-            validated = PassportService().verify(token=token)
-            check_csrf_token(request=request, user_id=validated.get("user_id"))
-        except:
-            res = False
-        return {"logged_in": res}

api/controllers/console/explore/banner.py

@@ -1,43 +0,0 @@
-from flask import request
-from flask_restx import Resource
-
-from controllers.console import api
-from controllers.console.explore.wraps import explore_banner_enabled
-from extensions.ext_database import db
-from models.model import ExporleBanner
-
-
-class BannerApi(Resource):
-    """Resource for banner list."""
-
-    @explore_banner_enabled
-    def get(self):
-        """Get banner list."""
-        language = request.args.get("language", "en-US")
-
-        # Build base query for enabled banners
-        base_query = db.session.query(ExporleBanner).where(ExporleBanner.status == "enabled")
-        
-        # Try to get banners in the requested language
-        banners = base_query.where(ExporleBanner.language == language).order_by(ExporleBanner.sort).all()
-        
-        # Fallback to en-US if no banners found and language is not en-US
-        if not banners and language != "en-US":
-            banners = base_query.where(ExporleBanner.language == "en-US").order_by(ExporleBanner.sort).all()
-        # Convert banners to serializable format
-        result = []
-        for banner in banners:
-            banner_data = {
-                "id": banner.id,
-                "content": banner.content,  # Already parsed as JSON by SQLAlchemy
-                "link": banner.link,
-                "sort": banner.sort,
-                "status": banner.status,
-                "created_at": banner.created_at.isoformat() if banner.created_at else None,
-            }
-            result.append(banner_data)
-
-        return result
-
-
-api.add_resource(BannerApi, "/explore/banners")

api/controllers/console/explore/error.py

@@ -29,25 +29,3 @@ class AppAccessDeniedError(BaseHTTPException):
     error_code = "access_denied"
     description = "App access denied."
     code = 403
-
-
-class TrialAppNotAllowed(BaseHTTPException):
-    """*403* `Trial App Not Allowed`
-
-    Raise if the user has reached the trial app limit.
-    """
-
-    error_code = "trial_app_not_allowed"
-    code = 403
-    description = "the app is not allowed to be trial."
-
-
-class TrialAppLimitExceeded(BaseHTTPException):
-    """*403* `Trial App Limit Exceeded`
-
-    Raise if the user has exceeded the trial app limit.
-    """
-
-    error_code = "trial_app_limit_exceeded"
-    code = 403
-    description = "The user has exceeded the trial app limit."

api/controllers/console/explore/recommended_app.py

@@ -27,7 +27,6 @@ recommended_app_fields = {
     "category": fields.String,
     "position": fields.Integer,
     "is_listed": fields.Boolean,
-    "can_trial": fields.Boolean,
 }
 
 recommended_app_list_fields = {

api/controllers/console/explore/trial.py

@@ -1,514 +0,0 @@
-import logging
-from typing import Any, cast
-
-from flask import request
-from flask_restx import Resource, marshal, marshal_with, reqparse
-from werkzeug.exceptions import Forbidden, InternalServerError, NotFound
-
-import services
-from controllers.common import fields
-from controllers.common.fields import build_site_model
-from controllers.console import api
-from controllers.console.app.error import (
-    AppUnavailableError,
-    AudioTooLargeError,
-    CompletionRequestError,
-    ConversationCompletedError,
-    NeedAddIdsError,
-    NoAudioUploadedError,
-    ProviderModelCurrentlyNotSupportError,
-    ProviderNotInitializeError,
-    ProviderNotSupportSpeechToTextError,
-    ProviderQuotaExceededError,
-    UnsupportedAudioTypeError,
-)
-from controllers.console.app.wraps import get_app_model_with_trial
-from controllers.console.explore.error import (
-    AppSuggestedQuestionsAfterAnswerDisabledError,
-    NotChatAppError,
-    NotCompletionAppError,
-    NotWorkflowAppError,
-)
-from controllers.console.explore.wraps import TrialAppResource, trial_feature_enable
-from controllers.service_api import service_api_ns
-from controllers.web.error import InvokeRateLimitError as InvokeRateLimitHttpError
-from core.app.app_config.common.parameters_mapping import get_parameters_from_feature_dict
-from core.app.apps.base_app_queue_manager import AppQueueManager
-from core.app.entities.app_invoke_entities import InvokeFrom
-from core.errors.error import (
-    ModelCurrentlyNotSupportError,
-    ProviderTokenNotInitError,
-    QuotaExceededError,
-)
-from core.model_runtime.errors.invoke import InvokeError
-from core.workflow.graph_engine.manager import GraphEngineManager
-from extensions.ext_database import db
-from fields.app_fields import app_detail_fields_with_site
-from fields.dataset_fields import dataset_fields
-from fields.workflow_fields import workflow_fields
-from libs import helper
-from libs.helper import uuid_value
-from libs.login import current_user
-from models import Account
-from models.account import TenantStatus
-from models.model import AppMode, Site
-from models.workflow import Workflow
-from services.app_generate_service import AppGenerateService
-from services.app_service import AppService
-from services.audio_service import AudioService
-from services.dataset_service import DatasetService
-from services.errors.audio import (
-    AudioTooLargeServiceError,
-    NoAudioUploadedServiceError,
-    ProviderNotSupportSpeechToTextServiceError,
-    UnsupportedAudioTypeServiceError,
-)
-from services.errors.conversation import ConversationNotExistsError
-from services.errors.llm import InvokeRateLimitError
-from services.errors.message import (
-    MessageNotExistsError,
-    SuggestedQuestionsAfterAnswerDisabledError,
-)
-from services.message_service import MessageService
-from services.recommended_app_service import RecommendedAppService
-
-logger = logging.getLogger(__name__)
-
-
-class TrialAppWorkflowRunApi(TrialAppResource):
-    def post(self, trial_app):
-        """
-        Run workflow
-        """
-        app_model = trial_app
-        if not app_model:
-            raise NotWorkflowAppError()
-        app_mode = AppMode.value_of(app_model.mode)
-        if app_mode != AppMode.WORKFLOW:
-            raise NotWorkflowAppError()
-
-        parser = reqparse.RequestParser()
-        parser.add_argument("inputs", type=dict, required=True, nullable=False, location="json")
-        parser.add_argument("files", type=list, required=False, location="json")
-        args = parser.parse_args()
-        assert current_user is not None
-        try:
-            app_id = app_model.id
-            user_id = current_user.id
-            response = AppGenerateService.generate(
-                app_model=app_model, user=current_user, args=args, invoke_from=InvokeFrom.EXPLORE, streaming=True
-            )
-            RecommendedAppService.add_trial_app_record(app_id, user_id)
-            return helper.compact_generate_response(response)
-        except ProviderTokenNotInitError as ex:
-            raise ProviderNotInitializeError(ex.description)
-        except QuotaExceededError:
-            raise ProviderQuotaExceededError()
-        except ModelCurrentlyNotSupportError:
-            raise ProviderModelCurrentlyNotSupportError()
-        except InvokeError as e:
-            raise CompletionRequestError(e.description)
-        except InvokeRateLimitError as ex:
-            raise InvokeRateLimitHttpError(ex.description)
-        except ValueError as e:
-            raise e
-        except Exception:
-            logger.exception("internal server error.")
-            raise InternalServerError()
-
-
-class TrialAppWorkflowTaskStopApi(TrialAppResource):
-    def post(self, trial_app, task_id: str):
-        """
-        Stop workflow task
-        """
-        app_model = trial_app
-        if not app_model:
-            raise NotWorkflowAppError()
-        app_mode = AppMode.value_of(app_model.mode)
-        if app_mode != AppMode.WORKFLOW:
-            raise NotWorkflowAppError()
-        assert current_user is not None
-
-        # Stop using both mechanisms for backward compatibility
-        # Legacy stop flag mechanism (without user check)
-        AppQueueManager.set_stop_flag_no_user_check(task_id)
-
-        # New graph engine command channel mechanism
-        GraphEngineManager.send_stop_command(task_id)
-
-        return {"result": "success"}
-
-
-class TrialChatApi(TrialAppResource):
-    @trial_feature_enable
-    def post(self, trial_app):
-        app_model = trial_app
-        app_mode = AppMode.value_of(app_model.mode)
-        if app_mode not in {AppMode.CHAT, AppMode.AGENT_CHAT, AppMode.ADVANCED_CHAT}:
-            raise NotChatAppError()
-
-        parser = reqparse.RequestParser()
-        parser.add_argument("inputs", type=dict, required=True, location="json")
-        parser.add_argument("query", type=str, required=True, location="json")
-        parser.add_argument("files", type=list, required=False, location="json")
-        parser.add_argument("conversation_id", type=uuid_value, location="json")
-        parser.add_argument("parent_message_id", type=uuid_value, required=False, location="json")
-        parser.add_argument("retriever_from", type=str, required=False, default="explore_app", location="json")
-        args = parser.parse_args()
-
-        args["auto_generate_name"] = False
-
-        try:
-            if not isinstance(current_user, Account):
-                raise ValueError("current_user must be an Account instance")
-            
-            # Get IDs before they might be detached from session
-            app_id = app_model.id
-            user_id = current_user.id
-            
-            response = AppGenerateService.generate(
-                app_model=app_model, user=current_user, args=args, invoke_from=InvokeFrom.EXPLORE, streaming=True
-            )
-            RecommendedAppService.add_trial_app_record(app_id, user_id)
-            return helper.compact_generate_response(response)
-        except services.errors.conversation.ConversationNotExistsError:
-            raise NotFound("Conversation Not Exists.")
-        except services.errors.conversation.ConversationCompletedError:
-            raise ConversationCompletedError()
-        except services.errors.app_model_config.AppModelConfigBrokenError:
-            logger.exception("App model config broken.")
-            raise AppUnavailableError()
-        except ProviderTokenNotInitError as ex:
-            raise ProviderNotInitializeError(ex.description)
-        except QuotaExceededError:
-            raise ProviderQuotaExceededError()
-        except ModelCurrentlyNotSupportError:
-            raise ProviderModelCurrentlyNotSupportError()
-        except InvokeError as e:
-            raise CompletionRequestError(e.description)
-        except InvokeRateLimitError as ex:
-            raise InvokeRateLimitHttpError(ex.description)
-        except ValueError as e:
-            raise e
-        except Exception:
-            logger.exception("internal server error.")
-            raise InternalServerError()
-
-
-class TrialMessageSuggestedQuestionApi(TrialAppResource):
-    @trial_feature_enable
-    def get(self, trial_app, message_id):
-        app_model = trial_app
-        app_mode = AppMode.value_of(app_model.mode)
-        if app_mode not in {AppMode.CHAT, AppMode.AGENT_CHAT, AppMode.ADVANCED_CHAT}:
-            raise NotChatAppError()
-
-        message_id = str(message_id)
-
-        try:
-            if not isinstance(current_user, Account):
-                raise ValueError("current_user must be an Account instance")
-            questions = MessageService.get_suggested_questions_after_answer(
-                app_model=app_model, user=current_user, message_id=message_id, invoke_from=InvokeFrom.EXPLORE
-            )
-        except MessageNotExistsError:
-            raise NotFound("Message not found")
-        except ConversationNotExistsError:
-            raise NotFound("Conversation not found")
-        except SuggestedQuestionsAfterAnswerDisabledError:
-            raise AppSuggestedQuestionsAfterAnswerDisabledError()
-        except ProviderTokenNotInitError as ex:
-            raise ProviderNotInitializeError(ex.description)
-        except QuotaExceededError:
-            raise ProviderQuotaExceededError()
-        except ModelCurrentlyNotSupportError:
-            raise ProviderModelCurrentlyNotSupportError()
-        except InvokeError as e:
-            raise CompletionRequestError(e.description)
-        except Exception:
-            logger.exception("internal server error.")
-            raise InternalServerError()
-
-        return {"data": questions}
-
-
-class TrialChatAudioApi(TrialAppResource):
-    @trial_feature_enable
-    def post(self, trial_app):
-        app_model = trial_app
-
-        file = request.files["file"]
-
-        try:
-            if not isinstance(current_user, Account):
-                raise ValueError("current_user must be an Account instance")
-            
-            # Get IDs before they might be detached from session
-            app_id = app_model.id
-            user_id = current_user.id
-            
-            response = AudioService.transcript_asr(app_model=app_model, file=file, end_user=None)
-            RecommendedAppService.add_trial_app_record(app_id, user_id)
-            return response
-        except services.errors.app_model_config.AppModelConfigBrokenError:
-            logger.exception("App model config broken.")
-            raise AppUnavailableError()
-        except NoAudioUploadedServiceError:
-            raise NoAudioUploadedError()
-        except AudioTooLargeServiceError as e:
-            raise AudioTooLargeError(str(e))
-        except UnsupportedAudioTypeServiceError:
-            raise UnsupportedAudioTypeError()
-        except ProviderNotSupportSpeechToTextServiceError:
-            raise ProviderNotSupportSpeechToTextError()
-        except ProviderTokenNotInitError as ex:
-            raise ProviderNotInitializeError(ex.description)
-        except QuotaExceededError:
-            raise ProviderQuotaExceededError()
-        except ModelCurrentlyNotSupportError:
-            raise ProviderModelCurrentlyNotSupportError()
-        except InvokeError as e:
-            raise CompletionRequestError(e.description)
-        except ValueError as e:
-            raise e
-        except Exception as e:
-            logger.exception("internal server error.")
-            raise InternalServerError()
-
-
-class TrialChatTextApi(TrialAppResource):
-    @trial_feature_enable
-    def post(self, trial_app):
-        app_model = trial_app
-        try:
-            parser = reqparse.RequestParser()
-            parser.add_argument("message_id", type=str, required=False, location="json")
-            parser.add_argument("voice", type=str, location="json")
-            parser.add_argument("text", type=str, location="json")
-            parser.add_argument("streaming", type=bool, location="json")
-            args = parser.parse_args()
-
-            message_id = args.get("message_id", None)
-            text = args.get("text", None)
-            voice = args.get("voice", None)
-            if not isinstance(current_user, Account):
-                raise ValueError("current_user must be an Account instance")
-            
-            # Get IDs before they might be detached from session
-            app_id = app_model.id
-            user_id = current_user.id
-            
-            response = AudioService.transcript_tts(app_model=app_model, text=text, voice=voice, message_id=message_id)
-            RecommendedAppService.add_trial_app_record(app_id, user_id)
-            return response
-        except services.errors.app_model_config.AppModelConfigBrokenError:
-            logger.exception("App model config broken.")
-            raise AppUnavailableError()
-        except NoAudioUploadedServiceError:
-            raise NoAudioUploadedError()
-        except AudioTooLargeServiceError as e:
-            raise AudioTooLargeError(str(e))
-        except UnsupportedAudioTypeServiceError:
-            raise UnsupportedAudioTypeError()
-        except ProviderNotSupportSpeechToTextServiceError:
-            raise ProviderNotSupportSpeechToTextError()
-        except ProviderTokenNotInitError as ex:
-            raise ProviderNotInitializeError(ex.description)
-        except QuotaExceededError:
-            raise ProviderQuotaExceededError()
-        except ModelCurrentlyNotSupportError:
-            raise ProviderModelCurrentlyNotSupportError()
-        except InvokeError as e:
-            raise CompletionRequestError(e.description)
-        except ValueError as e:
-            raise e
-        except Exception as e:
-            logger.exception("internal server error.")
-            raise InternalServerError()
-
-
-class TrialCompletionApi(TrialAppResource):
-    @trial_feature_enable
-    def post(self, trial_app):
-        app_model = trial_app
-        if app_model.mode != "completion":
-            raise NotCompletionAppError()
-
-        parser = reqparse.RequestParser()
-        parser.add_argument("inputs", type=dict, required=True, location="json")
-        parser.add_argument("query", type=str, location="json", default="")
-        parser.add_argument("files", type=list, required=False, location="json")
-        parser.add_argument("response_mode", type=str, choices=["blocking", "streaming"], location="json")
-        parser.add_argument("retriever_from", type=str, required=False, default="explore_app", location="json")
-        args = parser.parse_args()
-
-        streaming = args["response_mode"] == "streaming"
-        args["auto_generate_name"] = False
-
-        try:
-            if not isinstance(current_user, Account):
-                raise ValueError("current_user must be an Account instance")
-            
-            # Get IDs before they might be detached from session
-            app_id = app_model.id
-            user_id = current_user.id
-            
-            response = AppGenerateService.generate(
-                app_model=app_model, user=current_user, args=args, invoke_from=InvokeFrom.EXPLORE, streaming=streaming
-            )
-
-            RecommendedAppService.add_trial_app_record(app_id, user_id)
-            return helper.compact_generate_response(response)
-        except services.errors.conversation.ConversationNotExistsError:
-            raise NotFound("Conversation Not Exists.")
-        except services.errors.conversation.ConversationCompletedError:
-            raise ConversationCompletedError()
-        except services.errors.app_model_config.AppModelConfigBrokenError:
-            logger.exception("App model config broken.")
-            raise AppUnavailableError()
-        except ProviderTokenNotInitError as ex:
-            raise ProviderNotInitializeError(ex.description)
-        except QuotaExceededError:
-            raise ProviderQuotaExceededError()
-        except ModelCurrentlyNotSupportError:
-            raise ProviderModelCurrentlyNotSupportError()
-        except InvokeError as e:
-            raise CompletionRequestError(e.description)
-        except ValueError as e:
-            raise e
-        except Exception:
-            logger.exception("internal server error.")
-            raise InternalServerError()
-
-
-class TrialSitApi(Resource):
-    """Resource for trial app sites."""
-
-    @trial_feature_enable
-    @get_app_model_with_trial
-    @service_api_ns.marshal_with(build_site_model(service_api_ns))
-    def get(self, app_model):
-        """Retrieve app site info.
-
-        Returns the site configuration for the application including theme, icons, and text.
-        """
-        site = db.session.query(Site).where(Site.app_id == app_model.id).first()
-
-        if not site:
-            raise Forbidden()
-
-        assert app_model.tenant
-        if app_model.tenant.status == TenantStatus.ARCHIVE:
-            raise Forbidden()
-
-        return site
-
-
-class TrialAppParameterApi(Resource):
-    """Resource for app variables."""
-
-    @trial_feature_enable
-    @get_app_model_with_trial
-    @marshal_with(fields.parameters_fields)
-    def get(self, app_model):
-        """Retrieve app parameters."""
-
-        if app_model is None:
-            raise AppUnavailableError()
-
-        if app_model.mode in {AppMode.ADVANCED_CHAT, AppMode.WORKFLOW}:
-            workflow = app_model.workflow
-            if workflow is None:
-                raise AppUnavailableError()
-
-            features_dict = workflow.features_dict
-            user_input_form = workflow.user_input_form(to_old_structure=True)
-        else:
-            app_model_config = app_model.app_model_config
-            if app_model_config is None:
-                raise AppUnavailableError()
-
-            features_dict = app_model_config.to_dict()
-
-            user_input_form = features_dict.get("user_input_form", [])
-
-        return get_parameters_from_feature_dict(features_dict=features_dict, user_input_form=user_input_form)
-
-
-class AppApi(Resource):
-    @trial_feature_enable
-    @get_app_model_with_trial
-    @marshal_with(app_detail_fields_with_site)
-    def get(self, app_model):
-        """Get app detail"""
-
-        app_service = AppService()
-        app_model = app_service.get_app(app_model)
-
-        return app_model
-
-
-class AppWorkflowApi(Resource):
-    @trial_feature_enable
-    @get_app_model_with_trial
-    @marshal_with(workflow_fields)
-    def get(self, app_model):
-        """Get workflow detail"""
-        if not app_model.workflow_id:
-            raise AppUnavailableError()
-
-        workflow = (
-            db.session.query(Workflow)
-            .where(
-                Workflow.id == app_model.workflow_id,
-            )
-            .first()
-        )
-        return workflow
-
-
-class DatasetListApi(Resource):
-    @trial_feature_enable
-    @get_app_model_with_trial
-    def get(self, app_model):
-        page = request.args.get("page", default=1, type=int)
-        limit = request.args.get("limit", default=20, type=int)
-        ids = request.args.getlist("ids")
-      
-        tenant_id = app_model.tenant_id
-        if ids:
-            datasets, total = DatasetService.get_datasets_by_ids(ids, tenant_id)
-        else:
-            raise NeedAddIdsError()
-
-        data = cast(list[dict[str, Any]], marshal(datasets, dataset_fields))
-
-        response = {"data": data, "has_more": len(datasets) == limit, "limit": limit, "total": total, "page": page}
-        return response
-
-
-api.add_resource(TrialChatApi, "/trial-apps/<uuid:app_id>/chat-messages", endpoint="trial_app_chat_completion")
-
-api.add_resource(
-    TrialMessageSuggestedQuestionApi,
-    "/trial-apps/<uuid:app_id>/messages/<uuid:message_id>/suggested-questions",
-    endpoint="trial_app_suggested_question",
-)
-
-api.add_resource(TrialChatAudioApi, "/trial-apps/<uuid:app_id>/audio-to-text", endpoint="trial_app_audio")
-api.add_resource(TrialChatTextApi, "/trial-apps/<uuid:app_id>/text-to-audio", endpoint="trial_app_text")
-
-api.add_resource(TrialCompletionApi, "/trial-apps/<uuid:app_id>/completion-messages", endpoint="trial_app_completion")
-
-api.add_resource(TrialSitApi, "/trial-apps/<uuid:app_id>/site")
-
-api.add_resource(TrialAppParameterApi, "/trial-apps/<uuid:app_id>/parameters", endpoint="trial_app_parameters")
-
-api.add_resource(AppApi, "/trial-apps/<uuid:app_id>", endpoint="trial_app")
-
-api.add_resource(TrialAppWorkflowRunApi, "/trial-apps/<uuid:app_id>/workflows/run", endpoint="trial_app_workflow_run")
-api.add_resource(TrialAppWorkflowTaskStopApi, "/trial-apps/<uuid:app_id>/workflows/tasks/<string:task_id>/stop")
-
-api.add_resource(AppWorkflowApi, "/trial-apps/<uuid:app_id>/workflows", endpoint="trial_app_workflow")
-api.add_resource(DatasetListApi, "/trial-apps/<uuid:app_id>/datasets", endpoint="trial_app_datasets")

api/controllers/console/explore/wraps.py

@@ -2,16 +2,14 @@ from collections.abc import Callable
 from functools import wraps
 from typing import Concatenate, ParamSpec, TypeVar
 
-from flask import abort
-from flask_login import current_user
 from flask_restx import Resource
 from werkzeug.exceptions import NotFound
 
-from controllers.console.explore.error import AppAccessDeniedError, TrialAppLimitExceeded, TrialAppNotAllowed
+from controllers.console.explore.error import AppAccessDeniedError
 from controllers.console.wraps import account_initialization_required
 from extensions.ext_database import db
 from libs.login import current_account_with_tenant, login_required
-from models import AccountTrialAppRecord, App, InstalledApp, TrialApp
+from models import InstalledApp
 from services.enterprise.enterprise_service import EnterpriseService
 from services.feature_service import FeatureService
 
@@ -73,59 +71,6 @@ def user_allowed_to_access_app(view: Callable[Concatenate[InstalledApp, P], R] |
     return decorator
 
 
-def trial_app_required(view: Callable[Concatenate[App, P], R] | None = None):
-    def decorator(view: Callable[Concatenate[App, P], R]):
-        @wraps(view)
-        def decorated(app_id: str, *args: P.args, **kwargs: P.kwargs):
-            trial_app = db.session.query(TrialApp).where(TrialApp.app_id == str(app_id)).first()
-
-            if trial_app is None:
-                raise TrialAppNotAllowed()
-            app = trial_app.app
-
-            if app is None:
-                raise TrialAppNotAllowed()
-
-            account_trial_app_record = (
-                db.session.query(AccountTrialAppRecord)
-                .where(AccountTrialAppRecord.account_id == current_user.id, AccountTrialAppRecord.app_id == app_id)
-                .first()
-            )
-            if account_trial_app_record:
-                if account_trial_app_record.count >= trial_app.trial_limit:
-                    raise TrialAppLimitExceeded()
-
-            return view(app, *args, **kwargs)
-
-        return decorated
-
-    if view:
-        return decorator(view)
-    return decorator
-
-
-def trial_feature_enable(view: Callable[..., R]) -> Callable[..., R]:
-    @wraps(view)
-    def decorated(*args, **kwargs):
-        features = FeatureService.get_system_features()
-        if not features.enable_trial_app:
-            abort(403, "Trial app feature is not enabled.")
-        return view(*args, **kwargs)
-
-    return decorated
-
-
-def explore_banner_enabled(view: Callable[..., R]) -> Callable[..., R]:
-    @wraps(view)
-    def decorated(*args, **kwargs):
-        features = FeatureService.get_system_features()
-        if not features.enable_explore_banner:
-            abort(403, "Explore banner feature is not enabled.")
-        return view(*args, **kwargs)
-
-    return decorated
-
-
 class InstalledAppResource(Resource):
     # must be reversed if there are multiple decorators
 
@@ -135,13 +80,3 @@ class InstalledAppResource(Resource):
         account_initialization_required,
         login_required,
     ]
-
-
-class TrialAppResource(Resource):
-    # must be reversed if there are multiple decorators
-
-    method_decorators = [
-        trial_app_required,
-        account_initialization_required,
-        login_required,
-    ]

api/controllers/console/workspace/account.py

@@ -32,7 +32,6 @@ from controllers.console.wraps import (
     only_edition_cloud,
     setup_required,
 )
-from core.file import helpers as file_helpers
 from extensions.ext_database import db
 from fields.member_fields import account_fields
 from libs.datetime_utils import naive_utc_now
@@ -129,17 +128,6 @@ class AccountNameApi(Resource):
 
 @console_ns.route("/account/avatar")
 class AccountAvatarApi(Resource):
-    @setup_required
-    @login_required
-    @account_initialization_required
-    def get(self):
-        parser = reqparse.RequestParser()
-        parser.add_argument("avatar", type=str, required=True, location="args")
-        args = parser.parse_args()
-
-        avatar_url = file_helpers.get_signed_file_url(args["avatar"])
-        return {"avatar_url": avatar_url}
-
     @setup_required
     @login_required
     @account_initialization_required

api/controllers/console/workspace/workspace.py

@@ -51,9 +51,6 @@ tenant_fields = {
     "in_trial": fields.Boolean,
     "trial_end_reason": fields.String,
     "custom_config": fields.Raw(attribute="custom_config"),
-    "trial_credits": fields.Integer,
-    "trial_credits_used": fields.Integer,
-    "next_credit_reset_date": fields.Integer,
 }
 
 tenants_fields = {

api/controllers/service_api/__init__.py

@@ -19,7 +19,6 @@ from .app import (
     annotation,
     app,
     audio,
-    chatflow_memory,
     completion,
     conversation,
     file,
@@ -41,7 +40,6 @@ __all__ = [
     "annotation",
     "app",
     "audio",
-    "chatflow_memory",
     "completion",
     "conversation",
     "dataset",

api/controllers/service_api/app/chatflow_memory.py

@@ -1,109 +0,0 @@
-from flask_restx import Resource, reqparse
-
-from controllers.service_api import api
-from controllers.service_api.wraps import FetchUserArg, WhereisUserArg, validate_app_token
-from core.memory.entities import MemoryBlock, MemoryCreatedBy
-from core.workflow.runtime.variable_pool import VariablePool
-from models import App, EndUser
-from services.chatflow_memory_service import ChatflowMemoryService
-from services.workflow_service import WorkflowService
-
-
-class MemoryListApi(Resource):
-    @validate_app_token(fetch_user_arg=FetchUserArg(fetch_from=WhereisUserArg.JSON, required=True))
-    def get(self, app_model: App, end_user: EndUser):
-        parser = reqparse.RequestParser()
-        parser.add_argument("conversation_id", required=False, type=str | None, default=None)
-        parser.add_argument("memory_id", required=False, type=str | None, default=None)
-        parser.add_argument("version", required=False, type=int | None, default=None)
-        args = parser.parse_args()
-        conversation_id: str | None = args.get("conversation_id")
-        memory_id = args.get("memory_id")
-        version = args.get("version")
-
-        if conversation_id:
-            result = ChatflowMemoryService.get_persistent_memories_with_conversation(
-                app_model, MemoryCreatedBy(end_user_id=end_user.id), conversation_id, version
-            )
-            session_memories = ChatflowMemoryService.get_session_memories_with_conversation(
-                app_model, MemoryCreatedBy(end_user_id=end_user.id), conversation_id, version
-            )
-            result = [*result, *session_memories]
-        else:
-            result = ChatflowMemoryService.get_persistent_memories(
-                app_model, MemoryCreatedBy(end_user_id=end_user.id), version
-            )
-
-        if memory_id:
-            result = [it for it in result if it.spec.id == memory_id]
-        return [it for it in result if it.spec.end_user_visible]
-
-
-class MemoryEditApi(Resource):
-    @validate_app_token(fetch_user_arg=FetchUserArg(fetch_from=WhereisUserArg.JSON, required=True))
-    def put(self, app_model: App, end_user: EndUser):
-        parser = reqparse.RequestParser()
-        parser.add_argument("id", type=str, required=True)
-        parser.add_argument("conversation_id", type=str | None, required=False, default=None)
-        parser.add_argument("node_id", type=str | None, required=False, default=None)
-        parser.add_argument("update", type=str, required=True)
-        args = parser.parse_args()
-        workflow = WorkflowService().get_published_workflow(app_model)
-        update = args.get("update")
-        conversation_id = args.get("conversation_id")
-        node_id = args.get("node_id")
-        if not isinstance(update, str):
-            return {"error": "Invalid update"}, 400
-        if not workflow:
-            return {"error": "Workflow not found"}, 404
-        memory_spec = next((it for it in workflow.memory_blocks if it.id == args["id"]), None)
-        if not memory_spec:
-            return {"error": "Memory not found"}, 404
-
-        # First get existing memory
-        existing_memory = ChatflowMemoryService.get_memory_by_spec(
-            spec=memory_spec,
-            tenant_id=app_model.tenant_id,
-            app_id=app_model.id,
-            created_by=MemoryCreatedBy(end_user_id=end_user.id),
-            conversation_id=conversation_id,
-            node_id=node_id,
-            is_draft=False,
-        )
-
-        # Create updated memory instance with incremented version
-        updated_memory = MemoryBlock(
-            spec=existing_memory.spec,
-            tenant_id=existing_memory.tenant_id,
-            app_id=existing_memory.app_id,
-            conversation_id=existing_memory.conversation_id,
-            node_id=existing_memory.node_id,
-            value=update,  # New value
-            version=existing_memory.version + 1,  # Increment version for update
-            edited_by_user=True,
-            created_by=existing_memory.created_by,
-        )
-
-        ChatflowMemoryService.save_memory(updated_memory, VariablePool(), False)
-        return "", 204
-
-
-class MemoryDeleteApi(Resource):
-    @validate_app_token(fetch_user_arg=FetchUserArg(fetch_from=WhereisUserArg.JSON, required=True))
-    def delete(self, app_model: App, end_user: EndUser):
-        parser = reqparse.RequestParser()
-        parser.add_argument("id", type=str, required=False, default=None)
-        args = parser.parse_args()
-        memory_id = args.get("id")
-
-        if memory_id:
-            ChatflowMemoryService.delete_memory(app_model, memory_id, MemoryCreatedBy(end_user_id=end_user.id))
-            return "", 204
-        else:
-            ChatflowMemoryService.delete_all_user_memories(app_model, MemoryCreatedBy(end_user_id=end_user.id))
-            return "", 200
-
-
-api.add_resource(MemoryListApi, "/memories")
-api.add_resource(MemoryEditApi, "/memory-edit")
-api.add_resource(MemoryDeleteApi, "/memories")

api/controllers/web/__init__.py

@@ -18,7 +18,6 @@ web_ns = Namespace("web", description="Web application API operations", path="/"
 from . import (
     app,
     audio,
-    chatflow_memory,
     completion,
     conversation,
     feature,
@@ -40,7 +39,6 @@ __all__ = [
     "app",
     "audio",
     "bp",
-    "chatflow_memory",
     "completion",
     "conversation",
     "feature",

api/controllers/web/chatflow_memory.py

@@ -1,108 +0,0 @@
-from flask_restx import reqparse
-
-from controllers.web import api
-from controllers.web.wraps import WebApiResource
-from core.memory.entities import MemoryBlock, MemoryCreatedBy
-from core.workflow.runtime.variable_pool import VariablePool
-from models import App, EndUser
-from services.chatflow_memory_service import ChatflowMemoryService
-from services.workflow_service import WorkflowService
-
-
-class MemoryListApi(WebApiResource):
-    def get(self, app_model: App, end_user: EndUser):
-        parser = reqparse.RequestParser()
-        parser.add_argument("conversation_id", required=False, type=str | None, default=None)
-        parser.add_argument("memory_id", required=False, type=str | None, default=None)
-        parser.add_argument("version", required=False, type=int | None, default=None)
-        args = parser.parse_args()
-        conversation_id: str | None = args.get("conversation_id")
-        memory_id = args.get("memory_id")
-        version = args.get("version")
-
-        if conversation_id:
-            result = ChatflowMemoryService.get_persistent_memories_with_conversation(
-                app_model, MemoryCreatedBy(end_user_id=end_user.id), conversation_id, version
-            )
-            session_memories = ChatflowMemoryService.get_session_memories_with_conversation(
-                app_model, MemoryCreatedBy(end_user_id=end_user.id), conversation_id, version
-            )
-            result = [*result, *session_memories]
-        else:
-            result = ChatflowMemoryService.get_persistent_memories(
-                app_model, MemoryCreatedBy(end_user_id=end_user.id), version
-            )
-
-        if memory_id:
-            result = [it for it in result if it.spec.id == memory_id]
-        return [it for it in result if it.spec.end_user_visible]
-
-
-class MemoryEditApi(WebApiResource):
-    def put(self, app_model: App, end_user: EndUser):
-        parser = reqparse.RequestParser()
-        parser.add_argument("id", type=str, required=True)
-        parser.add_argument("conversation_id", type=str | None, required=False, default=None)
-        parser.add_argument("node_id", type=str | None, required=False, default=None)
-        parser.add_argument("update", type=str, required=True)
-        args = parser.parse_args()
-        workflow = WorkflowService().get_published_workflow(app_model)
-        update = args.get("update")
-        conversation_id = args.get("conversation_id")
-        node_id = args.get("node_id")
-        if not isinstance(update, str):
-            return {"error": "Update must be a string"}, 400
-        if not workflow:
-            return {"error": "Workflow not found"}, 404
-        memory_spec = next((it for it in workflow.memory_blocks if it.id == args["id"]), None)
-        if not memory_spec:
-            return {"error": "Memory not found"}, 404
-        if not memory_spec.end_user_editable:
-            return {"error": "Memory not editable"}, 403
-
-        # First get existing memory
-        existing_memory = ChatflowMemoryService.get_memory_by_spec(
-            spec=memory_spec,
-            tenant_id=app_model.tenant_id,
-            app_id=app_model.id,
-            created_by=MemoryCreatedBy(end_user_id=end_user.id),
-            conversation_id=conversation_id,
-            node_id=node_id,
-            is_draft=False,
-        )
-
-        # Create updated memory instance with incremented version
-        updated_memory = MemoryBlock(
-            spec=existing_memory.spec,
-            tenant_id=existing_memory.tenant_id,
-            app_id=existing_memory.app_id,
-            conversation_id=existing_memory.conversation_id,
-            node_id=existing_memory.node_id,
-            value=update,  # New value
-            version=existing_memory.version + 1,  # Increment version for update
-            edited_by_user=True,
-            created_by=existing_memory.created_by,
-        )
-
-        ChatflowMemoryService.save_memory(updated_memory, VariablePool(), False)
-        return "", 204
-
-
-class MemoryDeleteApi(WebApiResource):
-    def delete(self, app_model: App, end_user: EndUser):
-        parser = reqparse.RequestParser()
-        parser.add_argument("id", type=str, required=False, default=None)
-        args = parser.parse_args()
-        memory_id = args.get("id")
-
-        if memory_id:
-            ChatflowMemoryService.delete_memory(app_model, memory_id, MemoryCreatedBy(end_user_id=end_user.id))
-            return "", 204
-        else:
-            ChatflowMemoryService.delete_all_user_memories(app_model, MemoryCreatedBy(end_user_id=end_user.id))
-            return "", 200
-
-
-api.add_resource(MemoryListApi, "/memories")
-api.add_resource(MemoryEditApi, "/memory-edit")
-api.add_resource(MemoryDeleteApi, "/memories")

api/core/app/apps/advanced_chat/app_runner.py

@@ -1,11 +1,10 @@
 import logging
 import time
-from collections.abc import Mapping, MutableMapping, Sequence
+from collections.abc import Mapping, Sequence
 from typing import Any, cast
 
 from sqlalchemy import select
 from sqlalchemy.orm import Session
-from typing_extensions import override
 
 from core.app.apps.advanced_chat.app_config_manager import AdvancedChatAppConfig
 from core.app.apps.base_app_queue_manager import AppQueueManager
@@ -21,8 +20,6 @@ from core.app.entities.queue_entities import (
     QueueTextChunkEvent,
 )
 from core.app.features.annotation_reply.annotation_reply import AnnotationReplyFeature
-from core.memory.entities import MemoryCreatedBy, MemoryScope
-from core.model_runtime.entities import AssistantPromptMessage, UserPromptMessage
 from core.moderation.base import ModerationError
 from core.moderation.input_moderation import InputModeration
 from core.variables.variables import VariableUnion
@@ -30,7 +27,6 @@ from core.workflow.enums import WorkflowType
 from core.workflow.graph_engine.command_channels.redis_channel import RedisChannel
 from core.workflow.graph_engine.layers.base import GraphEngineLayer
 from core.workflow.graph_engine.layers.persistence import PersistenceWorkflowInfo, WorkflowPersistenceLayer
-from core.workflow.graph_events import GraphRunSucceededEvent
 from core.workflow.repositories.workflow_execution_repository import WorkflowExecutionRepository
 from core.workflow.repositories.workflow_node_execution_repository import WorkflowNodeExecutionRepository
 from core.workflow.runtime import GraphRuntimeState, VariablePool
@@ -43,8 +39,6 @@ from models import Workflow
 from models.enums import UserFrom
 from models.model import App, Conversation, Message, MessageAnnotation
 from models.workflow import ConversationVariable
-from services.chatflow_history_service import ChatflowHistoryService
-from services.chatflow_memory_service import ChatflowMemoryService
 
 logger = logging.getLogger(__name__)
 
@@ -87,10 +81,6 @@ class AdvancedChatAppRunner(WorkflowBasedAppRunner):
         self._workflow_node_execution_repository = workflow_node_execution_repository
 
     def run(self):
-        ChatflowMemoryService.wait_for_sync_memory_completion(
-            workflow=self._workflow, conversation_id=self.conversation.id
-        )
-
         app_config = self.application_generate_entity.app_config
         app_config = cast(AdvancedChatAppConfig, app_config)
 
@@ -153,7 +143,6 @@ class AdvancedChatAppRunner(WorkflowBasedAppRunner):
                 # Based on the definition of `VariableUnion`,
                 # `list[Variable]` can be safely used as `list[VariableUnion]` since they are compatible.
                 conversation_variables=conversation_variables,
-                memory_blocks=self._fetch_memory_blocks(),
             )
 
             # init graph
@@ -217,31 +206,6 @@ class AdvancedChatAppRunner(WorkflowBasedAppRunner):
         for event in generator:
             self._handle_event(workflow_entry, event)
 
-        try:
-            self._check_app_memory_updates(variable_pool)
-        except Exception as e:
-            logger.exception("Failed to check app memory updates", exc_info=e)
-
-    @override
-    def _handle_event(self, workflow_entry: WorkflowEntry, event: Any) -> None:
-        super()._handle_event(workflow_entry, event)
-        if isinstance(event, GraphRunSucceededEvent):
-            workflow_outputs = event.outputs
-            if not workflow_outputs:
-                logger.warning("Chatflow output is empty.")
-                return
-            assistant_message = workflow_outputs.get("answer")
-            if not assistant_message:
-                logger.warning("Chatflow output does not contain 'answer'.")
-                return
-            if not isinstance(assistant_message, str):
-                logger.warning("Chatflow output 'answer' is not a string.")
-                return
-            try:
-                self._sync_conversation_to_chatflow_tables(assistant_message)
-            except Exception as e:
-                logger.exception("Failed to sync conversation to memory tables", exc_info=e)
-
     def handle_input_moderation(
         self,
         app_record: App,
@@ -439,67 +403,3 @@ class AdvancedChatAppRunner(WorkflowBasedAppRunner):
 
         # Return combined list
         return existing_variables + new_variables
-
-    def _fetch_memory_blocks(self) -> Mapping[str, str]:
-        """fetch all memory blocks for current app"""
-
-        memory_blocks_dict: MutableMapping[str, str] = {}
-        is_draft = self.application_generate_entity.invoke_from == InvokeFrom.DEBUGGER
-        conversation_id = self.conversation.id
-        memory_block_specs = self._workflow.memory_blocks
-        # Get runtime memory values
-        memories = ChatflowMemoryService.get_memories_by_specs(
-            memory_block_specs=memory_block_specs,
-            tenant_id=self._workflow.tenant_id,
-            app_id=self._workflow.app_id,
-            node_id=None,
-            conversation_id=conversation_id,
-            is_draft=is_draft,
-            created_by=self._get_created_by(),
-        )
-
-        # Build memory_id -> value mapping
-        for memory in memories:
-            if memory.spec.scope == MemoryScope.APP:
-                # App level: use memory_id directly
-                memory_blocks_dict[memory.spec.id] = memory.value
-            else:  # NODE scope
-                node_id = memory.node_id
-                if not node_id:
-                    logger.warning("Memory block %s has no node_id, skip.", memory.spec.id)
-                    continue
-                key = f"{node_id}.{memory.spec.id}"
-                memory_blocks_dict[key] = memory.value
-
-        return memory_blocks_dict
-
-    def _sync_conversation_to_chatflow_tables(self, assistant_message: str):
-        ChatflowHistoryService.save_app_message(
-            prompt_message=UserPromptMessage(content=(self.application_generate_entity.query)),
-            conversation_id=self.conversation.id,
-            app_id=self._workflow.app_id,
-            tenant_id=self._workflow.tenant_id,
-        )
-        ChatflowHistoryService.save_app_message(
-            prompt_message=AssistantPromptMessage(content=assistant_message),
-            conversation_id=self.conversation.id,
-            app_id=self._workflow.app_id,
-            tenant_id=self._workflow.tenant_id,
-        )
-
-    def _check_app_memory_updates(self, variable_pool: VariablePool):
-        is_draft = self.application_generate_entity.invoke_from == InvokeFrom.DEBUGGER
-
-        ChatflowMemoryService.update_app_memory_if_needed(
-            workflow=self._workflow,
-            conversation_id=self.conversation.id,
-            variable_pool=variable_pool,
-            is_draft=is_draft,
-            created_by=self._get_created_by(),
-        )
-
-    def _get_created_by(self) -> MemoryCreatedBy:
-        if self.application_generate_entity.invoke_from in {InvokeFrom.DEBUGGER, InvokeFrom.EXPLORE}:
-            return MemoryCreatedBy(account_id=self.application_generate_entity.user_id)
-        else:
-            return MemoryCreatedBy(end_user_id=self.application_generate_entity.user_id)

api/core/app/apps/pipeline/pipeline_generator.py

@@ -40,7 +40,6 @@ from core.workflow.repositories.draft_variable_repository import DraftVariableSa
 from core.workflow.repositories.workflow_execution_repository import WorkflowExecutionRepository
 from core.workflow.repositories.workflow_node_execution_repository import WorkflowNodeExecutionRepository
 from core.workflow.variable_loader import DUMMY_VARIABLE_LOADER, VariableLoader
-from enums.cloud_plan import CloudPlan
 from extensions.ext_database import db
 from libs.flask_utils import preserve_flask_contexts
 from models import Account, EndUser, Workflow, WorkflowNodeExecutionTriggeredFrom
@@ -48,6 +47,7 @@ from models.dataset import Document, DocumentPipelineExecutionLog, Pipeline
 from models.enums import WorkflowRunTriggeredFrom
 from models.model import AppMode
 from services.datasource_provider_service import DatasourceProviderService
+from services.rag_pipeline.rag_pipeline_task_proxy import RagPipelineTaskProxy
 from services.workflow_draft_variable_service import DraftVarLoader, WorkflowDraftVariableService
 
 logger = logging.getLogger(__name__)
@@ -244,34 +244,7 @@ class PipelineGenerator(BaseAppGenerator):
                 )
 
         if rag_pipeline_invoke_entities:
-            # store the rag_pipeline_invoke_entities to object storage
-            text = [item.model_dump() for item in rag_pipeline_invoke_entities]
-            name = "rag_pipeline_invoke_entities.json"
-            # Convert list to proper JSON string
-            json_text = json.dumps(text)
-            upload_file = FileService(db.engine).upload_text(json_text, name, user.id, dataset.tenant_id)
-            features = FeatureService.get_features(dataset.tenant_id)
-            if features.billing.enabled and features.billing.subscription.plan == CloudPlan.SANDBOX:
-                tenant_pipeline_task_key = f"tenant_pipeline_task:{dataset.tenant_id}"
-                tenant_self_pipeline_task_queue = f"tenant_self_pipeline_task_queue:{dataset.tenant_id}"
-
-                if redis_client.get(tenant_pipeline_task_key):
-                    # Add to waiting queue using List operations (lpush)
-                    redis_client.lpush(tenant_self_pipeline_task_queue, upload_file.id)
-                else:
-                    # Set flag and execute task
-                    redis_client.set(tenant_pipeline_task_key, 1, ex=60 * 60)
-                    rag_pipeline_run_task.delay(  # type: ignore
-                        rag_pipeline_invoke_entities_file_id=upload_file.id,
-                        tenant_id=dataset.tenant_id,
-                    )
-
-            else:
-                priority_rag_pipeline_run_task.delay(  # type: ignore
-                    rag_pipeline_invoke_entities_file_id=upload_file.id,
-                    tenant_id=dataset.tenant_id,
-                )
-
+            RagPipelineTaskProxy(dataset.tenant_id, user.id, rag_pipeline_invoke_entities).delay()
         # return batch, dataset, documents
         return {
             "batch": batch,

api/core/entities/document_task.py

@@ -1,13 +1,15 @@
+from collections.abc import Sequence
 from dataclasses import dataclass
 
 
 @dataclass
 class DocumentTask:
     """Document task entity for document indexing operations.
-    
+
     This class represents a document indexing task that can be queued
     and processed by the document indexing system.
     """
+
     tenant_id: str
     dataset_id: str
-    document_ids: list[str]
+    document_ids: Sequence[str]

api/core/entities/mcp_provider.py

@@ -14,7 +14,6 @@ from core.helper.provider_cache import NoOpProviderCredentialCache
 from core.mcp.types import OAuthClientInformation, OAuthClientMetadata, OAuthTokens
 from core.tools.entities.common_entities import I18nObject
 from core.tools.entities.tool_entities import ToolProviderType
-from core.tools.utils.encryption import create_provider_encrypter
 
 if TYPE_CHECKING:
     from models.tools import MCPToolProvider
@@ -272,6 +271,7 @@ class MCPProviderEntity(BaseModel):
 
     def _decrypt_dict(self, data: dict[str, Any]) -> dict[str, Any]:
         """Generic method to decrypt dictionary fields"""
+        from core.tools.utils.encryption import create_provider_encrypter
 
         if not data:
             return {}

api/core/entities/provider_configuration.py

@@ -1533,6 +1533,9 @@ class ProviderConfiguration(BaseModel):
             # Return composite sort key: (model_type value, model position index)
             return (model.model_type.value, position_index)
 
+        # Deduplicate
+        provider_models = list({(m.model, m.model_type, m.fetch_from): m for m in provider_models}.values())
+
         # Sort using the composite sort key
         return sorted(provider_models, key=get_sort_key)
 

api/core/hosting_configuration.py

@@ -56,9 +56,6 @@ class HostingConfiguration:
         self.provider_map[f"{DEFAULT_PLUGIN_ID}/minimax/minimax"] = self.init_minimax()
         self.provider_map[f"{DEFAULT_PLUGIN_ID}/spark/spark"] = self.init_spark()
         self.provider_map[f"{DEFAULT_PLUGIN_ID}/zhipuai/zhipuai"] = self.init_zhipuai()
-        self.provider_map[f"{DEFAULT_PLUGIN_ID}/gemini/google"] = self.init_gemini()
-        self.provider_map[f"{DEFAULT_PLUGIN_ID}/x/x"] = self.init_xai()
-        self.provider_map[f"{DEFAULT_PLUGIN_ID}/deepseek/deepseek"] = self.init_deepseek()
 
         self.moderation_config = self.init_moderation_config()
 
@@ -131,7 +128,7 @@ class HostingConfiguration:
         quotas: list[HostingQuota] = []
 
         if dify_config.HOSTED_OPENAI_TRIAL_ENABLED:
-            hosted_quota_limit = 0
+            hosted_quota_limit = dify_config.HOSTED_OPENAI_QUOTA_LIMIT
             trial_models = self.parse_restrict_models_from_env("HOSTED_OPENAI_TRIAL_MODELS")
             trial_quota = TrialHostingQuota(quota_limit=hosted_quota_limit, restrict_models=trial_models)
             quotas.append(trial_quota)
@@ -159,49 +156,18 @@ class HostingConfiguration:
             quota_unit=quota_unit,
         )
 
-    def init_gemini(self) -> HostingProvider:
-        quota_unit = QuotaUnit.CREDITS
-        quotas: list[HostingQuota] = []
-
-        if dify_config.HOSTED_GEMINI_TRIAL_ENABLED:
-            hosted_quota_limit = 0
-            trial_models = self.parse_restrict_models_from_env("HOSTED_GEMINI_TRIAL_MODELS")
-            trial_quota = TrialHostingQuota(quota_limit=hosted_quota_limit, restrict_models=trial_models)
-            quotas.append(trial_quota)
-
-        if dify_config.HOSTED_GEMINI_PAID_ENABLED:
-            paid_models = self.parse_restrict_models_from_env("HOSTED_GEMINI_PAID_MODELS")
-            paid_quota = PaidHostingQuota(restrict_models=paid_models)
-            quotas.append(paid_quota)
-
-        if len(quotas) > 0:
-            credentials = {
-                "google_api_key": dify_config.HOSTED_GEMINI_API_KEY,
-            }
-
-            if dify_config.HOSTED_GEMINI_API_BASE:
-                credentials["google_base_url"] = dify_config.HOSTED_GEMINI_API_BASE
-
-            return HostingProvider(enabled=True, credentials=credentials, quota_unit=quota_unit, quotas=quotas)
-
-        return HostingProvider(
-            enabled=False,
-            quota_unit=quota_unit,
-        )
-
-    def init_anthropic(self) -> HostingProvider:
-        quota_unit = QuotaUnit.CREDITS
+    @staticmethod
+    def init_anthropic() -> HostingProvider:
+        quota_unit = QuotaUnit.TOKENS
         quotas: list[HostingQuota] = []
 
         if dify_config.HOSTED_ANTHROPIC_TRIAL_ENABLED:
-            hosted_quota_limit = 0
-            trail_models = self.parse_restrict_models_from_env("HOSTED_ANTHROPIC_TRIAL_MODELS")
-            trial_quota = TrialHostingQuota(quota_limit=hosted_quota_limit, restrict_models=trail_models)
+            hosted_quota_limit = dify_config.HOSTED_ANTHROPIC_QUOTA_LIMIT
+            trial_quota = TrialHostingQuota(quota_limit=hosted_quota_limit)
             quotas.append(trial_quota)
 
         if dify_config.HOSTED_ANTHROPIC_PAID_ENABLED:
-            paid_models = self.parse_restrict_models_from_env("HOSTED_ANTHROPIC_PAID_MODELS")
-            paid_quota = PaidHostingQuota(restrict_models=paid_models)
+            paid_quota = PaidHostingQuota()
             quotas.append(paid_quota)
 
         if len(quotas) > 0:
@@ -219,66 +185,6 @@ class HostingConfiguration:
             quota_unit=quota_unit,
         )
 
-    def init_xai(self) -> HostingProvider:
-        quota_unit = QuotaUnit.CREDITS
-        quotas: list[HostingQuota] = []
-
-        if dify_config.HOSTED_XAI_TRIAL_ENABLED:
-            hosted_quota_limit = 0
-            trail_models = self.parse_restrict_models_from_env("HOSTED_XAI_TRIAL_MODELS")
-            trial_quota = TrialHostingQuota(quota_limit=hosted_quota_limit, restrict_models=trail_models)
-            quotas.append(trial_quota)
-
-        if dify_config.HOSTED_XAI_PAID_ENABLED:
-            paid_models = self.parse_restrict_models_from_env("HOSTED_XAI_PAID_MODELS")
-            paid_quota = PaidHostingQuota(restrict_models=paid_models)
-            quotas.append(paid_quota)
-
-        if len(quotas) > 0:
-            credentials = {
-                "api_key": dify_config.HOSTED_XAI_API_KEY,
-            }
-
-            if dify_config.HOSTED_XAI_API_BASE:
-                credentials["endpoint_url"] = dify_config.HOSTED_XAI_API_BASE
-
-            return HostingProvider(enabled=True, credentials=credentials, quota_unit=quota_unit, quotas=quotas)
-
-        return HostingProvider(
-            enabled=False,
-            quota_unit=quota_unit,
-        )
-
-    def init_deepseek(self) -> HostingProvider:
-        quota_unit = QuotaUnit.CREDITS
-        quotas: list[HostingQuota] = []
-
-        if dify_config.HOSTED_DEEPSEEK_TRIAL_ENABLED:
-            hosted_quota_limit = 0
-            trail_models = self.parse_restrict_models_from_env("HOSTED_DEEPSEEK_TRIAL_MODELS")
-            trial_quota = TrialHostingQuota(quota_limit=hosted_quota_limit, restrict_models=trail_models)
-            quotas.append(trial_quota)
-
-        if dify_config.HOSTED_DEEPSEEK_PAID_ENABLED:
-            paid_models = self.parse_restrict_models_from_env("HOSTED_DEEPSEEK_PAID_MODELS")
-            paid_quota = PaidHostingQuota(restrict_models=paid_models)
-            quotas.append(paid_quota)
-
-        if len(quotas) > 0:
-            credentials = {
-                "api_key": dify_config.HOSTED_DEEPSEEK_API_KEY,
-            }
-
-            if dify_config.HOSTED_DEEPSEEK_API_BASE:
-                credentials["endpoint_url"] = dify_config.HOSTED_DEEPSEEK_API_BASE
-
-            return HostingProvider(enabled=True, credentials=credentials, quota_unit=quota_unit, quotas=quotas)
-
-        return HostingProvider(
-            enabled=False,
-            quota_unit=quota_unit,
-        )
-
     @staticmethod
     def init_minimax() -> HostingProvider:
         quota_unit = QuotaUnit.TOKENS

api/core/llm_generator/llm_generator.py

@@ -14,12 +14,10 @@ from core.llm_generator.prompts import (
     JAVASCRIPT_CODE_GENERATOR_PROMPT_TEMPLATE,
     LLM_MODIFY_CODE_SYSTEM,
     LLM_MODIFY_PROMPT_SYSTEM,
-    MEMORY_UPDATE_PROMPT,
     PYTHON_CODE_GENERATOR_PROMPT_TEMPLATE,
     SYSTEM_STRUCTURED_OUTPUT_GENERATE,
     WORKFLOW_RULE_CONFIG_PROMPT_GENERATE_TEMPLATE,
 )
-from core.memory.entities import MemoryBlock, MemoryBlockSpec
 from core.model_manager import ModelManager
 from core.model_runtime.entities.llm_entities import LLMResult
 from core.model_runtime.entities.message_entities import PromptMessage, SystemPromptMessage, UserPromptMessage
@@ -30,7 +28,6 @@ from core.ops.ops_trace_manager import TraceQueueManager, TraceTask
 from core.ops.utils import measure_time
 from core.prompt.utils.prompt_template_parser import PromptTemplateParser
 from core.workflow.entities.workflow_node_execution import WorkflowNodeExecutionMetadataKey
-from core.workflow.runtime.variable_pool import VariablePool
 from extensions.ext_database import db
 from extensions.ext_storage import storage
 from models import App, Message, WorkflowNodeExecutionModel
@@ -565,35 +562,3 @@ class LLMGenerator:
                 "Failed to invoke LLM model, model: %s", json.dumps(model_config.get("name")), exc_info=True
             )
             return {"error": f"An unexpected error occurred: {str(e)}"}
-
-    @staticmethod
-    def update_memory_block(
-        tenant_id: str,
-        visible_history: Sequence[tuple[str, str]],
-        variable_pool: VariablePool,
-        memory_block: MemoryBlock,
-        memory_spec: MemoryBlockSpec,
-    ) -> str:
-        model_instance = ModelManager().get_model_instance(
-            tenant_id=tenant_id,
-            provider=memory_spec.model.provider,
-            model=memory_spec.model.name,
-            model_type=ModelType.LLM,
-        )
-        formatted_history = ""
-        for sender, message in visible_history:
-            formatted_history += f"{sender}: {message}\n"
-        filled_instruction = variable_pool.convert_template(memory_spec.instruction).text
-        formatted_prompt = PromptTemplateParser(MEMORY_UPDATE_PROMPT).format(
-            inputs={
-                "formatted_history": formatted_history,
-                "current_value": memory_block.value,
-                "instruction": filled_instruction,
-            }
-        )
-        llm_result = model_instance.invoke_llm(
-            prompt_messages=[UserPromptMessage(content=formatted_prompt)],
-            model_parameters=memory_spec.model.completion_params,
-            stream=False,
-        )
-        return llm_result.message.get_text_content()

api/core/llm_generator/prompts.py

@@ -422,18 +422,3 @@ INSTRUCTION_GENERATE_TEMPLATE_PROMPT = """The output of this prompt is not as ex
 You should edit the prompt according to the IDEAL OUTPUT."""
 
 INSTRUCTION_GENERATE_TEMPLATE_CODE = """Please fix the errors in the {{#error_message#}}."""
-
-MEMORY_UPDATE_PROMPT = """
-Based on the following conversation history, update the memory content:
-
-Conversation history:
-{{formatted_history}}
-
-Current memory:
-{{current_value}}
-
-Update instruction:
-{{instruction}}
-
-Please output only the updated memory content, no other text like greeting:
-"""

api/core/memory/entities.py

@@ -1,119 +0,0 @@
-from __future__ import annotations
-
-from enum import StrEnum
-from typing import TYPE_CHECKING, Optional
-from uuid import uuid4
-
-from pydantic import BaseModel, Field
-
-if TYPE_CHECKING:
-    from core.app.app_config.entities import ModelConfig
-
-
-class MemoryScope(StrEnum):
-    """Memory scope determined by node_id field"""
-    APP = "app"  # node_id is None
-    NODE = "node"  # node_id is not None
-
-
-class MemoryTerm(StrEnum):
-    """Memory term determined by conversation_id field"""
-    SESSION = "session"  # conversation_id is not None
-    PERSISTENT = "persistent"  # conversation_id is None
-
-
-class MemoryStrategy(StrEnum):
-    ON_TURNS = "on_turns"
-
-
-class MemoryScheduleMode(StrEnum):
-    SYNC = "sync"
-    ASYNC = "async"
-
-
-class MemoryBlockSpec(BaseModel):
-    """Memory block specification for workflow configuration"""
-    id: str = Field(
-        default_factory=lambda: str(uuid4()),
-        description="Unique identifier for the memory block",
-    )
-    name: str = Field(description="Display name of the memory block")
-    description: str = Field(default="", description="Description of the memory block")
-    template: str = Field(description="Initial template content for the memory")
-    instruction: str = Field(description="Instructions for updating the memory")
-    scope: MemoryScope = Field(description="Scope of the memory (app or node level)")
-    term: MemoryTerm = Field(description="Term of the memory (session or persistent)")
-    strategy: MemoryStrategy = Field(description="Update strategy for the memory")
-    update_turns: int = Field(gt=0, description="Number of turns between updates")
-    preserved_turns: int = Field(gt=0, description="Number of conversation turns to preserve")
-    schedule_mode: MemoryScheduleMode = Field(description="Synchronous or asynchronous update mode")
-    model: ModelConfig = Field(description="Model configuration for memory updates")
-    end_user_visible: bool = Field(default=False, description="Whether memory is visible to end users")
-    end_user_editable: bool = Field(default=False, description="Whether memory is editable by end users")
-
-
-class MemoryCreatedBy(BaseModel):
-    end_user_id: str | None = None
-    account_id: str | None = None
-
-
-class MemoryBlock(BaseModel):
-    """Runtime memory block instance
-
-    Design Rules:
-    - app_id = None: Global memory (future feature, not implemented yet)
-    - app_id = str: App-specific memory
-    - conversation_id = None: Persistent memory (cross-conversation)
-    - conversation_id = str: Session memory (conversation-specific)
-    - node_id = None: App-level scope
-    - node_id = str: Node-level scope
-
-    These rules implicitly determine scope and term without redundant storage.
-    """
-    spec: MemoryBlockSpec
-    tenant_id: str
-    value: str
-    app_id: str
-    conversation_id: Optional[str] = None
-    node_id: Optional[str] = None
-    edited_by_user: bool = False
-    created_by: MemoryCreatedBy
-    version: int = Field(description="Memory block version number")
-
-
-class MemoryValueData(BaseModel):
-    value: str
-    edited_by_user: bool = False
-
-
-class ChatflowConversationMetadata(BaseModel):
-    """Metadata for chatflow conversation with visible message count"""
-    type: str = "mutable_visible_window"
-    visible_count: int = Field(gt=0, description="Number of visible messages to keep")
-
-
-class MemoryBlockWithConversation(MemoryBlock):
-    """MemoryBlock with optional conversation metadata for session memories"""
-    conversation_metadata: ChatflowConversationMetadata = Field(
-        description="Conversation metadata, only present for session memories"
-    )
-
-    @classmethod
-    def from_memory_block(
-        cls,
-        memory_block: MemoryBlock,
-        conversation_metadata: ChatflowConversationMetadata
-    ) -> MemoryBlockWithConversation:
-        """Create MemoryBlockWithConversation from MemoryBlock"""
-        return cls(
-            spec=memory_block.spec,
-            tenant_id=memory_block.tenant_id,
-            value=memory_block.value,
-            app_id=memory_block.app_id,
-            conversation_id=memory_block.conversation_id,
-            node_id=memory_block.node_id,
-            edited_by_user=memory_block.edited_by_user,
-            created_by=memory_block.created_by,
-            version=memory_block.version,
-            conversation_metadata=conversation_metadata
-        )

api/core/memory/errors.py

@@ -1,6 +0,0 @@
-class MemorySyncTimeoutError(Exception):
-    def __init__(self, app_id: str, conversation_id: str):
-        self.app_id = app_id
-        self.conversation_id = conversation_id
-        self.message = "Memory synchronization timeout after 50 seconds"
-        super().__init__(self.message)

api/core/prompt/entities/advanced_prompt_entities.py

@@ -45,12 +45,6 @@ class MemoryConfig(BaseModel):
         enabled: bool
         size: int | None = None
 
-    mode: Literal["linear", "block"] | None = "linear"
-    block_id: list[str] | None = None
     role_prefix: RolePrefix | None = None
     window: WindowConfig
     query_prompt_template: str | None = None
-
-    @property
-    def is_block_mode(self) -> bool:
-        return self.mode == "block" and bool(self.block_id)

api/core/provider_manager.py

@@ -618,18 +618,18 @@ class ProviderManager:
                 )
 
             for quota in configuration.quotas:
-                if quota.quota_type in (ProviderQuotaType.TRIAL, ProviderQuotaType.PAID):
+                if quota.quota_type == ProviderQuotaType.TRIAL:
                     # Init trial provider records if not exists
-                    if quota.quota_type not in provider_quota_to_provider_record_dict:
+                    if ProviderQuotaType.TRIAL not in provider_quota_to_provider_record_dict:
                         try:
                             # FIXME ignore the type error, only TrialHostingQuota has limit need to change the logic
                             new_provider_record = Provider(
                                 tenant_id=tenant_id,
                                 # TODO: Use provider name with prefix after the data migration.
                                 provider_name=ModelProviderID(provider_name).provider_name,
-                                provider_type=ProviderType.SYSTEM.value,
-                                quota_type=quota.quota_type,
-                                quota_limit=0,  # type: ignore
+                                provider_type=ProviderType.SYSTEM,
+                                quota_type=ProviderQuotaType.TRIAL,
+                                quota_limit=quota.quota_limit,  # type: ignore
                                 quota_used=0,
                                 is_valid=True,
                             )
@@ -641,8 +641,8 @@ class ProviderManager:
                             stmt = select(Provider).where(
                                 Provider.tenant_id == tenant_id,
                                 Provider.provider_name == ModelProviderID(provider_name).provider_name,
-                                Provider.provider_type == ProviderType.SYSTEM.value,
-                                Provider.quota_type == quota.quota_type,
+                                Provider.provider_type == ProviderType.SYSTEM,
+                                Provider.quota_type == ProviderQuotaType.TRIAL,
                             )
                             existed_provider_record = db.session.scalar(stmt)
                             if not existed_provider_record:
@@ -652,7 +652,7 @@ class ProviderManager:
                                 existed_provider_record.is_valid = True
                                 db.session.commit()
 
-                            provider_name_to_provider_records_dict[provider_name].append(existed_provider_record)                        
+                            provider_name_to_provider_records_dict[provider_name].append(existed_provider_record)
 
         return provider_name_to_provider_records_dict
 
@@ -912,22 +912,6 @@ class ProviderManager:
                 provider_record
             )
         quota_configurations = []
-
-        if dify_config.EDITION == "CLOUD":
-            from services.credit_pool_service import CreditPoolService
-
-            trail_pool = CreditPoolService.get_pool(
-                tenant_id=tenant_id,
-                pool_type=ProviderQuotaType.TRIAL.value,
-            )
-            paid_pool = CreditPoolService.get_pool(
-                tenant_id=tenant_id,
-                pool_type=ProviderQuotaType.PAID.value,
-            )
-        else:
-            trail_pool = None
-            paid_pool = None
-
         for provider_quota in provider_hosting_configuration.quotas:
             if provider_quota.quota_type not in quota_type_to_provider_records_dict:
                 if provider_quota.quota_type == ProviderQuotaType.FREE:
@@ -948,36 +932,16 @@ class ProviderManager:
                     raise ValueError("quota_used is None")
                 if provider_record.quota_limit is None:
                     raise ValueError("quota_limit is None")
-                if provider_quota.quota_type == ProviderQuotaType.TRIAL and trail_pool is not None:
-                    quota_configuration = QuotaConfiguration(
-                        quota_type=provider_quota.quota_type,
-                        quota_unit=provider_hosting_configuration.quota_unit or QuotaUnit.TOKENS,
-                        quota_used=trail_pool.quota_used,
-                        quota_limit=trail_pool.quota_limit,
-                        is_valid=trail_pool.quota_limit > trail_pool.quota_used or trail_pool.quota_limit == -1,
-                        restrict_models=provider_quota.restrict_models,
-                    )
 
-                elif provider_quota.quota_type == ProviderQuotaType.PAID and paid_pool is not None:
-                    quota_configuration = QuotaConfiguration(
-                        quota_type=provider_quota.quota_type,
-                        quota_unit=provider_hosting_configuration.quota_unit or QuotaUnit.TOKENS,
-                        quota_used=paid_pool.quota_used,
-                        quota_limit=paid_pool.quota_limit,
-                        is_valid=paid_pool.quota_limit > paid_pool.quota_used or paid_pool.quota_limit == -1,
-                        restrict_models=provider_quota.restrict_models,
-                    )
-
-                else:
-                    quota_configuration = QuotaConfiguration(
-                        quota_type=provider_quota.quota_type,
-                        quota_unit=provider_hosting_configuration.quota_unit or QuotaUnit.TOKENS,
-                        quota_used=provider_record.quota_used,
-                        quota_limit=provider_record.quota_limit,
-                        is_valid=provider_record.quota_limit > provider_record.quota_used
-                        or provider_record.quota_limit == -1,
-                        restrict_models=provider_quota.restrict_models,
-                    )
+                quota_configuration = QuotaConfiguration(
+                    quota_type=provider_quota.quota_type,
+                    quota_unit=provider_hosting_configuration.quota_unit or QuotaUnit.TOKENS,
+                    quota_used=provider_record.quota_used,
+                    quota_limit=provider_record.quota_limit,
+                    is_valid=provider_record.quota_limit > provider_record.quota_used
+                    or provider_record.quota_limit == -1,
+                    restrict_models=provider_quota.restrict_models,
+                )
 
             quota_configurations.append(quota_configuration)
 

api/core/rag/datasource/vdb/elasticsearch/elasticsearch_vector.py

@@ -147,7 +147,8 @@ class ElasticSearchVector(BaseVector):
 
     def _get_version(self) -> str:
         info = self._client.info()
-        return cast(str, info["version"]["number"])
+        # remove any suffix like "-SNAPSHOT" from the version string
+        return cast(str, info["version"]["number"]).split("-")[0]
 
     def _check_version(self):
         if parse_version(self._version) < parse_version("8.0.0"):

api/core/rag/datasource/vdb/oceanbase/oceanbase_vector.py

@@ -37,13 +37,13 @@ class OceanBaseVectorConfig(BaseModel):
     @classmethod
     def validate_config(cls, values: dict):
         if not values["host"]:
-            raise ValueError("config OCEANBASE_VECTOR_HOST is required")
+            raise ValueError("config OCEANBASE_HOST is required")
         if not values["port"]:
-            raise ValueError("config OCEANBASE_VECTOR_PORT is required")
+            raise ValueError("config OCEANBASE_PORT is required")
         if not values["user"]:
-            raise ValueError("config OCEANBASE_VECTOR_USER is required")
+            raise ValueError("config OCEANBASE_USER is required")
         if not values["database"]:
-            raise ValueError("config OCEANBASE_VECTOR_DATABASE is required")
+            raise ValueError("config OCEANBASE_DATABASE is required")
         return values
 
 
@@ -316,11 +316,11 @@ class OceanBaseVectorFactory(AbstractVectorFactory):
         return OceanBaseVector(
             collection_name,
             OceanBaseVectorConfig(
-                host=dify_config.OCEANBASE_VECTOR_HOST or "",
-                port=dify_config.OCEANBASE_VECTOR_PORT or 0,
-                user=dify_config.OCEANBASE_VECTOR_USER or "",
-                password=(dify_config.OCEANBASE_VECTOR_PASSWORD or ""),
-                database=dify_config.OCEANBASE_VECTOR_DATABASE or "",
+                host=dify_config.OCEANBASE_HOST or "",
+                port=dify_config.OCEANBASE_PORT or 0,
+                user=dify_config.OCEANBASE_USER or "",
+                password=(dify_config.OCEANBASE_PASSWORD or ""),
+                database=dify_config.OCEANBASE_DATABASE or "",
                 enable_hybrid_search=dify_config.OCEANBASE_ENABLE_HYBRID_SEARCH or False,
             ),
         )

api/core/rag/datasource/vdb/oracle/oraclevector.py

@@ -289,8 +289,7 @@ class OracleVector(BaseVector):
                 words = pseg.cut(query)
                 current_entity = ""
                 for word, pos in words:
-                    # nr: person name, ns: place name, nt: organization name
-                    if pos in {"nr", "Ng", "eng", "nz", "n", "ORG", "v"}:
+                    if pos in {"nr", "Ng", "eng", "nz", "n", "ORG", "v"}:  # nr: 人名，ns: 地名，nt: 机构名
                         current_entity += word
                     else:
                         if current_entity:

api/core/rag/datasource/vdb/pyvastbase/vastbase_vector.py

@@ -213,7 +213,7 @@ class VastbaseVector(BaseVector):
 
             with self._get_cursor() as cur:
                 cur.execute(SQL_CREATE_TABLE.format(table_name=self.table_name, dimension=dimension))
-                # Vastbase supports vector dimensions in range [1, 16000]
+                # Vastbase 支持的向量维度取值范围为 [1,16000]
                 if dimension <= 16000:
                     cur.execute(SQL_CREATE_INDEX.format(table_name=self.table_name))
             redis_client.set(collection_exist_cache_key, 1, ex=3600)

api/core/rag/datasource/vdb/weaviate/weaviate_vector.py

@@ -92,7 +92,7 @@ class WeaviateVector(BaseVector):
 
         # Parse gRPC configuration
         if config.grpc_endpoint:
-            # Urls without scheme won't be parsed correctly in some python verions,
+            # Urls without scheme won't be parsed correctly in some python versions,
             # see https://bugs.python.org/issue27657
             grpc_endpoint_with_scheme = (
                 config.grpc_endpoint if "://" in config.grpc_endpoint else f"grpc://{config.grpc_endpoint}"

api/core/rag/pipeline/queue.py

@@ -1,89 +1,79 @@
 import json
-from dataclasses import dataclass
+from collections.abc import Sequence
 from typing import Any
 
+from pydantic import BaseModel, ValidationError
+
 from extensions.ext_redis import redis_client
 
-TASK_WRAPPER_PREFIX = "__WRAPPER__:"
+_DEFAULT_TASK_TTL = 60 * 60  # 1 hour
 
 
-@dataclass
-class TaskWrapper:
+class TaskWrapper(BaseModel):
     data: Any
-    
+
     def serialize(self) -> str:
-        return json.dumps(self.data, ensure_ascii=False)
-    
+        return self.model_dump_json()
+
     @classmethod
-    def deserialize(cls, serialized_data: str) -> 'TaskWrapper':
-        data = json.loads(serialized_data)
-        return cls(data)
+    def deserialize(cls, serialized_data: str) -> "TaskWrapper":
+        return cls.model_validate_json(serialized_data)
 
 
-class TenantSelfTaskQueue:
+class TenantIsolatedTaskQueue:
     """
-    Simple queue for tenant self tasks, used for tenant self task isolation.
+    Simple queue for tenant isolated tasks, used for rag related tenant tasks isolation.
     It uses Redis list to store tasks, and Redis key to store task waiting flag.
     Support tasks that can be serialized by json.
     """
-    DEFAULT_TASK_TTL = 60 * 60
 
     def __init__(self, tenant_id: str, unique_key: str):
-        self.tenant_id = tenant_id
-        self.unique_key = unique_key
-        self.queue = f"tenant_self_{unique_key}_task_queue:{tenant_id}"
-        self.task_key = f"tenant_{unique_key}_task:{tenant_id}"
+        self._tenant_id = tenant_id
+        self._unique_key = unique_key
+        self._queue = f"tenant_self_{unique_key}_task_queue:{tenant_id}"
+        self._task_key = f"tenant_{unique_key}_task:{tenant_id}"
 
     def get_task_key(self):
-        return redis_client.get(self.task_key)
+        return redis_client.get(self._task_key)
 
-    def set_task_waiting_time(self, ttl: int | None = None):
-        ttl = ttl or self.DEFAULT_TASK_TTL
-        redis_client.setex(self.task_key, ttl, 1)
+    def set_task_waiting_time(self, ttl: int = _DEFAULT_TASK_TTL):
+        redis_client.setex(self._task_key, ttl, 1)
 
     def delete_task_key(self):
-        redis_client.delete(self.task_key)
+        redis_client.delete(self._task_key)
 
-    def push_tasks(self, tasks: list):
+    def push_tasks(self, tasks: Sequence[Any]):
         serialized_tasks = []
         for task in tasks:
             # Store str list directly, maintaining full compatibility for pipeline scenarios
             if isinstance(task, str):
                 serialized_tasks.append(task)
             else:
-                # Use TaskWrapper to do JSON serialization, add prefix for identification
-                wrapper = TaskWrapper(task)
+                # Use TaskWrapper to do JSON serialization for non-string tasks
+                wrapper = TaskWrapper(data=task)
                 serialized_data = wrapper.serialize()
-                serialized_tasks.append(f"{TASK_WRAPPER_PREFIX}{serialized_data}")
-        
-        redis_client.lpush(self.queue, *serialized_tasks)
-    
-    def pull_tasks(self, count: int = 1) -> list:
+                serialized_tasks.append(serialized_data)
+
+        redis_client.lpush(self._queue, *serialized_tasks)
+
+    def pull_tasks(self, count: int = 1) -> Sequence[Any]:
         if count <= 0:
             return []
-        
+
         tasks = []
         for _ in range(count):
-            serialized_task = redis_client.rpop(self.queue)
+            serialized_task = redis_client.rpop(self._queue)
             if not serialized_task:
                 break
-            
-            if isinstance(serialized_task, bytes):
-                serialized_task = serialized_task.decode('utf-8')
-            
-            # Check if use TaskWrapper or not
-            if serialized_task.startswith(TASK_WRAPPER_PREFIX):
-                try:
-                    wrapper_data = serialized_task[len(TASK_WRAPPER_PREFIX):]
-                    wrapper = TaskWrapper.deserialize(wrapper_data)
-                    tasks.append(wrapper.data)
-                except (json.JSONDecodeError, TypeError, ValueError):
-                    tasks.append(serialized_task)
-            else:
-                tasks.append(serialized_task)
-        
-        return tasks
 
-    def get_next_task(self):
-        tasks = self.pull_tasks(1)
-        return tasks[0] if tasks else None
+            if isinstance(serialized_task, bytes):
+                serialized_task = serialized_task.decode("utf-8")
+
+            try:
+                wrapper = TaskWrapper.deserialize(serialized_task)
+                tasks.append(wrapper.data)
+            except (json.JSONDecodeError, ValidationError, TypeError, ValueError):
+                # Fall back to raw string for legacy format or invalid JSON
+                tasks.append(serialized_task)
+
+        return tasks

api/core/tools/mcp_tool/tool.py

@@ -1,16 +1,19 @@
 import base64
 import json
+import logging
 from collections.abc import Generator
 from typing import Any
 
 from core.mcp.auth_client import MCPClientWithAuthRetry
 from core.mcp.error import MCPConnectionError
-from core.mcp.types import CallToolResult, ImageContent, TextContent
+from core.mcp.types import AudioContent, CallToolResult, ImageContent, TextContent
 from core.tools.__base.tool import Tool
 from core.tools.__base.tool_runtime import ToolRuntime
 from core.tools.entities.tool_entities import ToolEntity, ToolInvokeMessage, ToolProviderType
 from core.tools.errors import ToolInvokeError
 
+logger = logging.getLogger(__name__)
+
 
 class MCPTool(Tool):
     def __init__(
@@ -52,6 +55,11 @@ class MCPTool(Tool):
                 yield from self._process_text_content(content)
             elif isinstance(content, ImageContent):
                 yield self._process_image_content(content)
+            elif isinstance(content, AudioContent):
+                yield self._process_audio_content(content)
+            else:
+                logger.warning("Unsupported content type=%s", type(content))
+
         # handle MCP structured output
         if self.entity.output_schema and result.structuredContent:
             for k, v in result.structuredContent.items():
@@ -97,6 +105,10 @@ class MCPTool(Tool):
         """Process image content and return a blob message."""
         return self.create_blob_message(blob=base64.b64decode(content.data), meta={"mime_type": content.mimeType})
 
+    def _process_audio_content(self, content: AudioContent) -> ToolInvokeMessage:
+        """Process audio content and return a blob message."""
+        return self.create_blob_message(blob=base64.b64decode(content.data), meta={"mime_type": content.mimeType})
+
     def fork_tool_runtime(self, runtime: ToolRuntime) -> "MCPTool":
         return MCPTool(
             entity=self.entity,

api/core/tools/tool_manager.py

@@ -14,6 +14,7 @@ from sqlalchemy.orm import Session
 from yarl import URL
 
 import contexts
+from configs import dify_config
 from core.helper.provider_cache import ToolProviderCredentialsCache
 from core.plugin.impl.tool import PluginToolManager
 from core.tools.__base.tool_provider import ToolProviderController
@@ -33,7 +34,6 @@ from services.tools.mcp_tools_manage_service import MCPToolManageService
 if TYPE_CHECKING:
     from core.workflow.nodes.tool.entities import ToolEntity
 
-from configs import dify_config
 from core.agent.entities import AgentToolEntity
 from core.app.entities.app_invoke_entities import InvokeFrom
 from core.helper.module_import_helper import load_single_subclass_from_source
@@ -621,12 +621,28 @@ class ToolManager:
         """
         # according to multi credentials, select the one with is_default=True first, then created_at oldest
         # for compatibility with old version
-        sql = """
+        if dify_config.SQLALCHEMY_DATABASE_URI_SCHEME == "postgresql":
+            # PostgreSQL: Use DISTINCT ON
+            sql = """
                 SELECT DISTINCT ON (tenant_id, provider) id
                 FROM tool_builtin_providers
                 WHERE tenant_id = :tenant_id
                 ORDER BY tenant_id, provider, is_default DESC, created_at DESC
                 """
+        else:
+            # MySQL: Use window function to achieve same result
+            sql = """
+                SELECT id FROM (
+                    SELECT id, 
+                           ROW_NUMBER() OVER (
+                               PARTITION BY tenant_id, provider 
+                               ORDER BY is_default DESC, created_at DESC
+                           ) as rn
+                    FROM tool_builtin_providers
+                    WHERE tenant_id = :tenant_id
+                ) ranked WHERE rn = 1
+                """
+        
         with Session(db.engine, autoflush=False) as session:
             ids = [row.id for row in session.execute(sa.text(sql), {"tenant_id": tenant_id}).all()]
             return session.query(BuiltinToolProvider).where(BuiltinToolProvider.id.in_(ids)).all()

api/core/variables/segments.py

@@ -203,48 +203,6 @@ class ArrayFileSegment(ArraySegment):
         return ""
 
 
-class VersionedMemoryValue(BaseModel):
-    current_value: str = None  # type: ignore
-    versions: Mapping[str, str] = {}
-
-    model_config = ConfigDict(frozen=True)
-
-    def add_version(
-        self,
-        new_value: str,
-        version_name: str | None = None
-    ) -> "VersionedMemoryValue":
-        if version_name is None:
-            version_name = str(len(self.versions) + 1)
-        if version_name in self.versions:
-            raise ValueError(f"Version '{version_name}' already exists.")
-        self.current_value = new_value
-        return VersionedMemoryValue(
-            current_value=new_value,
-            versions={
-                version_name: new_value,
-                **self.versions,
-            }
-        )
-
-
-class VersionedMemorySegment(Segment):
-    value_type: SegmentType = SegmentType.VERSIONED_MEMORY
-    value: VersionedMemoryValue = None  # type: ignore
-
-    @property
-    def text(self) -> str:
-        return self.value.current_value
-
-    @property
-    def log(self) -> str:
-        return self.value.current_value
-
-    @property
-    def markdown(self) -> str:
-        return self.value.current_value
-
-
 class ArrayBooleanSegment(ArraySegment):
     value_type: SegmentType = SegmentType.ARRAY_BOOLEAN
     value: Sequence[bool]
@@ -290,7 +248,6 @@ SegmentUnion: TypeAlias = Annotated[
         | Annotated[ArrayObjectSegment, Tag(SegmentType.ARRAY_OBJECT)]
         | Annotated[ArrayFileSegment, Tag(SegmentType.ARRAY_FILE)]
         | Annotated[ArrayBooleanSegment, Tag(SegmentType.ARRAY_BOOLEAN)]
-        | Annotated[VersionedMemorySegment, Tag(SegmentType.VERSIONED_MEMORY)]
     ),
     Discriminator(get_segment_discriminator),
 ]

api/core/variables/types.py

@@ -41,8 +41,6 @@ class SegmentType(StrEnum):
     ARRAY_FILE = "array[file]"
     ARRAY_BOOLEAN = "array[boolean]"
 
-    VERSIONED_MEMORY = "versioned_memory"
-
     NONE = "none"
 
     GROUP = "group"

api/core/variables/variables.py

@@ -22,7 +22,6 @@ from .segments import (
     ObjectSegment,
     Segment,
     StringSegment,
-    VersionedMemorySegment,
     get_segment_discriminator,
 )
 from .types import SegmentType
@@ -107,10 +106,6 @@ class ArrayFileVariable(ArrayFileSegment, ArrayVariable):
     pass
 
 
-class VersionedMemoryVariable(VersionedMemorySegment, Variable):
-    pass
-
-
 class ArrayBooleanVariable(ArrayBooleanSegment, ArrayVariable):
     pass
 
@@ -166,7 +161,6 @@ VariableUnion: TypeAlias = Annotated[
         | Annotated[ArrayFileVariable, Tag(SegmentType.ARRAY_FILE)]
         | Annotated[ArrayBooleanVariable, Tag(SegmentType.ARRAY_BOOLEAN)]
         | Annotated[SecretVariable, Tag(SegmentType.SECRET)]
-        | Annotated[VersionedMemoryVariable, Tag(SegmentType.VERSIONED_MEMORY)]
     ),
     Discriminator(get_segment_discriminator),
 ]

api/core/workflow/constants.py

@@ -1,5 +1,4 @@
 SYSTEM_VARIABLE_NODE_ID = "sys"
 ENVIRONMENT_VARIABLE_NODE_ID = "env"
 CONVERSATION_VARIABLE_NODE_ID = "conversation"
-MEMORY_BLOCK_VARIABLE_NODE_ID = "memory_block"
 RAG_PIPELINE_VARIABLE_NODE_ID = "rag"

api/core/workflow/nodes/llm/llm_utils.py

@@ -6,7 +6,7 @@ from sqlalchemy.orm import Session
 
 from configs import dify_config
 from core.app.entities.app_invoke_entities import ModelConfigWithCredentialsEntity
-from core.entities.provider_entities import ProviderQuotaType, QuotaUnit
+from core.entities.provider_entities import QuotaUnit
 from core.file.models import File
 from core.memory.token_buffer_memory import TokenBufferMemory
 from core.model_manager import ModelInstance, ModelManager
@@ -136,36 +136,21 @@ def deduct_llm_quota(tenant_id: str, model_instance: ModelInstance, usage: LLMUs
             used_quota = 1
 
     if used_quota is not None and system_configuration.current_quota_type is not None:
-        if system_configuration.current_quota_type == ProviderQuotaType.TRIAL:
-            from services.credit_pool_service import CreditPoolService
-
-            CreditPoolService.check_and_deduct_credits(
-                tenant_id=tenant_id,
-                credits_required=used_quota,
-            )
-        elif system_configuration.current_quota_type == ProviderQuotaType.PAID:
-            from services.credit_pool_service import CreditPoolService
-            CreditPoolService.check_and_deduct_credits(
-                tenant_id=tenant_id,
-                credits_required=used_quota,
-                pool_type="paid",
-            )
-        else:
-            with Session(db.engine) as session:
-                stmt = (
-                    update(Provider)
-                    .where(
-                        Provider.tenant_id == tenant_id,
-                        # TODO: Use provider name with prefix after the data migration.
-                        Provider.provider_name == ModelProviderID(model_instance.provider).provider_name,
-                        Provider.provider_type == ProviderType.SYSTEM.value,
-                        Provider.quota_type == system_configuration.current_quota_type.value,
-                        Provider.quota_limit > Provider.quota_used,
-                    )
-                    .values(
-                        quota_used=Provider.quota_used + used_quota,
-                        last_used=naive_utc_now(),
-                    )
+        with Session(db.engine) as session:
+            stmt = (
+                update(Provider)
+                .where(
+                    Provider.tenant_id == tenant_id,
+                    # TODO: Use provider name with prefix after the data migration.
+                    Provider.provider_name == ModelProviderID(model_instance.provider).provider_name,
+                    Provider.provider_type == ProviderType.SYSTEM,
+                    Provider.quota_type == system_configuration.current_quota_type.value,
+                    Provider.quota_limit > Provider.quota_used,
                 )
-                session.execute(stmt)
-                session.commit()
+                .values(
+                    quota_used=Provider.quota_used + used_quota,
+                    last_used=naive_utc_now(),
+                )
+            )
+            session.execute(stmt)
+            session.commit()

api/core/workflow/nodes/llm/node.py

@@ -7,15 +7,11 @@ import time
 from collections.abc import Generator, Mapping, Sequence
 from typing import TYPE_CHECKING, Any, Literal
 
-from sqlalchemy import select
-from sqlalchemy.orm import Session
-
-from core.app.entities.app_invoke_entities import InvokeFrom, ModelConfigWithCredentialsEntity
+from core.app.entities.app_invoke_entities import ModelConfigWithCredentialsEntity
 from core.file import FileType, file_manager
 from core.helper.code_executor import CodeExecutor, CodeLanguage
 from core.llm_generator.output_parser.errors import OutputParserError
 from core.llm_generator.output_parser.structured_output import invoke_llm_with_structured_output
-from core.memory.entities import MemoryCreatedBy, MemoryScope
 from core.memory.token_buffer_memory import TokenBufferMemory
 from core.model_manager import ModelInstance, ModelManager
 from core.model_runtime.entities import (
@@ -77,8 +73,6 @@ from core.workflow.nodes.base.entities import BaseNodeData, RetryConfig, Variabl
 from core.workflow.nodes.base.node import Node
 from core.workflow.nodes.base.variable_template_parser import VariableTemplateParser
 from core.workflow.runtime import VariablePool
-from models import UserFrom, Workflow
-from models.engine import db
 
 from . import llm_utils
 from .entities import (
@@ -323,11 +317,6 @@ class LLMNode(Node):
             if self._file_outputs:
                 outputs["files"] = ArrayFileSegment(value=self._file_outputs)
 
-            try:
-                self._handle_chatflow_memory(result_text, variable_pool)
-            except Exception as e:
-                logger.warning("Memory orchestration failed for node %s: %s", self.node_id, str(e))
-
             # Send final chunk event to indicate streaming is complete
             yield StreamChunkEvent(
                 selector=[self._node_id, "text"],
@@ -1239,77 +1228,6 @@ class LLMNode(Node):
     def retry(self) -> bool:
         return self._node_data.retry_config.retry_enabled
 
-    def _handle_chatflow_memory(self, llm_output: str, variable_pool: VariablePool):
-        if not self._node_data.memory or self._node_data.memory.mode != "block":
-            return
-
-        conversation_id_segment = variable_pool.get((SYSTEM_VARIABLE_NODE_ID, SystemVariableKey.CONVERSATION_ID))
-        if not conversation_id_segment:
-            raise ValueError("Conversation ID not found in variable pool.")
-        conversation_id = conversation_id_segment.text
-
-        user_query_segment = variable_pool.get((SYSTEM_VARIABLE_NODE_ID, SystemVariableKey.QUERY))
-        if not user_query_segment:
-            raise ValueError("User query not found in variable pool.")
-        user_query = user_query_segment.text
-
-        from core.model_runtime.entities.message_entities import AssistantPromptMessage, UserPromptMessage
-        from services.chatflow_history_service import ChatflowHistoryService
-
-        ChatflowHistoryService.save_node_message(
-            prompt_message=(UserPromptMessage(content=user_query)),
-            node_id=self.node_id,
-            conversation_id=conversation_id,
-            app_id=self.app_id,
-            tenant_id=self.tenant_id,
-        )
-        ChatflowHistoryService.save_node_message(
-            prompt_message=(AssistantPromptMessage(content=llm_output)),
-            node_id=self.node_id,
-            conversation_id=conversation_id,
-            app_id=self.app_id,
-            tenant_id=self.tenant_id,
-        )
-
-        memory_config = self._node_data.memory
-        if not memory_config:
-            return
-        block_ids = memory_config.block_id
-        if not block_ids:
-            return
-
-        # FIXME: This is dirty workaround and may cause incorrect resolution for workflow version
-        with Session(db.engine) as session:
-            stmt = select(Workflow).where(Workflow.tenant_id == self.tenant_id, Workflow.app_id == self.app_id)
-            workflow = session.scalars(stmt).first()
-            if not workflow:
-                raise ValueError("Workflow not found.")
-        memory_blocks = workflow.memory_blocks
-
-        for block_id in block_ids:
-            memory_block_spec = next((block for block in memory_blocks if block.id == block_id), None)
-
-            if memory_block_spec and memory_block_spec.scope == MemoryScope.NODE:
-                is_draft = self.invoke_from == InvokeFrom.DEBUGGER
-                from services.chatflow_memory_service import ChatflowMemoryService
-
-                ChatflowMemoryService.update_node_memory_if_needed(
-                    tenant_id=self.tenant_id,
-                    app_id=self.app_id,
-                    node_id=self.id,
-                    conversation_id=conversation_id,
-                    memory_block_spec=memory_block_spec,
-                    variable_pool=variable_pool,
-                    is_draft=is_draft,
-                    created_by=self._get_user_from_context(),
-                )
-
-    def _get_user_from_context(self) -> MemoryCreatedBy:
-        if self.user_from == UserFrom.ACCOUNT:
-            return MemoryCreatedBy(account_id=self.user_id)
-        else:
-            return MemoryCreatedBy(end_user_id=self.user_id)
-
 
 def _combine_message_content_with_role(
     *, contents: str | list[PromptMessageContentUnionTypes] | None = None, role: PromptMessageRole

api/core/workflow/runtime/variable_pool.py

@@ -9,12 +9,11 @@ from pydantic import BaseModel, Field
 from core.file import File, FileAttribute, file_manager
 from core.variables import Segment, SegmentGroup, Variable
 from core.variables.consts import SELECTORS_LENGTH
-from core.variables.segments import FileSegment, ObjectSegment, VersionedMemoryValue
-from core.variables.variables import RAGPipelineVariableInput, VariableUnion, VersionedMemoryVariable
+from core.variables.segments import FileSegment, ObjectSegment
+from core.variables.variables import RAGPipelineVariableInput, VariableUnion
 from core.workflow.constants import (
     CONVERSATION_VARIABLE_NODE_ID,
     ENVIRONMENT_VARIABLE_NODE_ID,
-    MEMORY_BLOCK_VARIABLE_NODE_ID,
     RAG_PIPELINE_VARIABLE_NODE_ID,
     SYSTEM_VARIABLE_NODE_ID,
 )
@@ -57,10 +56,6 @@ class VariablePool(BaseModel):
         description="RAG pipeline variables.",
         default_factory=list,
     )
-    memory_blocks: Mapping[str, str] = Field(
-        description="Memory blocks.",
-        default_factory=dict,
-    )
 
     def model_post_init(self, context: Any, /):
         # Create a mapping from field names to SystemVariableKey enum values
@@ -81,18 +76,6 @@ class VariablePool(BaseModel):
                 rag_pipeline_variables_map[node_id][key] = value
             for key, value in rag_pipeline_variables_map.items():
                 self.add((RAG_PIPELINE_VARIABLE_NODE_ID, key), value)
-        # Add memory blocks to the variable pool
-        for memory_id, memory_value in self.memory_blocks.items():
-            self.add(
-                [MEMORY_BLOCK_VARIABLE_NODE_ID, memory_id],
-                VersionedMemoryVariable(
-                    value=VersionedMemoryValue(
-                        current_value=memory_value,
-                        versions={"1": memory_value},
-                    ),
-                    name=memory_id,
-                )
-            )
 
     def add(self, selector: Sequence[str], value: Any, /):
         """
@@ -170,7 +153,11 @@ class VariablePool(BaseModel):
             return None
 
         node_id, name = self._selector_to_keys(selector)
-        segment: Segment | None = self.variable_dictionary[node_id].get(name)
+        node_map = self.variable_dictionary.get(node_id)
+        if node_map is None:
+            return None
+
+        segment: Segment | None = node_map.get(name)
 
         if segment is None:
             return None

api/docker/entrypoint.sh

@@ -32,23 +32,21 @@ if [[ "${MODE}" == "worker" ]]; then
 
   exec celery -A celery_entrypoint.celery worker -P ${CELERY_WORKER_CLASS:-gevent} $CONCURRENCY_OPTION \
     --max-tasks-per-child ${MAX_TASKS_PER_CHILD:-50} --loglevel ${LOG_LEVEL:-INFO} \
-    -Q ${CELERY_QUEUES:-dataset,mail,ops_trace,app_deletion,plugin,workflow_storage,conversation,priority_pipeline,pipeline} \
+    -Q ${CELERY_QUEUES:-dataset,priority_dataset,mail,ops_trace,app_deletion,plugin,workflow_storage,conversation,priority_pipeline,pipeline} \
     --prefetch-multiplier=1
 
 elif [[ "${MODE}" == "beat" ]]; then
   exec celery -A app.celery beat --loglevel ${LOG_LEVEL:-INFO}
 else
   if [[ "${DEBUG}" == "true" ]]; then
-    export HOST=${DIFY_BIND_ADDRESS:-0.0.0.0}
-    export PORT=${DIFY_PORT:-5001}
-    exec python -m app
+    exec flask run --host=${DIFY_BIND_ADDRESS:-0.0.0.0} --port=${DIFY_PORT:-5001} --debug
   else
     exec gunicorn \
       --bind "${DIFY_BIND_ADDRESS:-0.0.0.0}:${DIFY_PORT:-5001}" \
       --workers ${SERVER_WORKER_AMOUNT:-1} \
-      --worker-class ${SERVER_WORKER_CLASS:-geventwebsocket.gunicorn.workers.GeventWebSocketWorker} \
+      --worker-class ${SERVER_WORKER_CLASS:-gevent} \
       --worker-connections ${SERVER_WORKER_CONNECTIONS:-10} \
       --timeout ${GUNICORN_TIMEOUT:-200} \
-      app:socketio_app
+      app:app
   fi
 fi

api/events/event_handlers/update_provider_when_message_created.py

@@ -10,7 +10,7 @@ from sqlalchemy.orm import Session
 
 from configs import dify_config
 from core.app.entities.app_invoke_entities import AgentChatAppGenerateEntity, ChatAppGenerateEntity
-from core.entities.provider_entities import ProviderQuotaType, QuotaUnit, SystemConfiguration
+from core.entities.provider_entities import QuotaUnit, SystemConfiguration
 from events.message_event import message_was_created
 from extensions.ext_database import db
 from extensions.ext_redis import redis_client, redis_fallback
@@ -134,38 +134,22 @@ def handle(sender: Message, **kwargs):
             system_configuration=system_configuration,
             model_name=model_config.model,
         )
+
         if used_quota is not None:
-            if provider_configuration.system_configuration.current_quota_type == ProviderQuotaType.TRIAL:
-                from services.credit_pool_service import CreditPoolService
-
-                CreditPoolService.check_and_deduct_credits(
+            quota_update = _ProviderUpdateOperation(
+                filters=_ProviderUpdateFilters(
                     tenant_id=tenant_id,
-                    credits_required=used_quota,
-                    pool_type="trial",
-                )
-            elif provider_configuration.system_configuration.current_quota_type == ProviderQuotaType.PAID:
-                from services.credit_pool_service import CreditPoolService
-
-                CreditPoolService.check_and_deduct_credits(
-                    tenant_id=tenant_id,
-                    credits_required=used_quota,
-                    pool_type="paid",
-                )
-            else:
-                quota_update = _ProviderUpdateOperation(
-                    filters=_ProviderUpdateFilters(
-                        tenant_id=tenant_id,
-                        provider_name=ModelProviderID(model_config.provider).provider_name,
-                        provider_type=ProviderType.SYSTEM.value,
-                        quota_type=provider_configuration.system_configuration.current_quota_type.value,
-                    ),
-                    values=_ProviderUpdateValues(quota_used=Provider.quota_used + used_quota, last_used=current_time),
-                    additional_filters=_ProviderUpdateAdditionalFilters(
-                        quota_limit_check=True  # Provider.quota_limit > Provider.quota_used
-                    ),
-                    description="quota_deduction_update",
-                )
-                updates_to_perform.append(quota_update)
+                    provider_name=ModelProviderID(model_config.provider).provider_name,
+                    provider_type=ProviderType.SYSTEM,
+                    quota_type=provider_configuration.system_configuration.current_quota_type.value,
+                ),
+                values=_ProviderUpdateValues(quota_used=Provider.quota_used + used_quota, last_used=current_time),
+                additional_filters=_ProviderUpdateAdditionalFilters(
+                    quota_limit_check=True  # Provider.quota_limit > Provider.quota_used
+                ),
+                description="quota_deduction_update",
+            )
+            updates_to_perform.append(quota_update)
 
     # Execute all updates
     start_time = time_module.perf_counter()

api/extensions/ext_blueprints.py

@@ -31,13 +31,7 @@ def init_app(app: DifyApp):
         web_bp,
         resources={r"/*": {"origins": dify_config.WEB_API_CORS_ALLOW_ORIGINS}},
         supports_credentials=True,
-        allow_headers=[
-            "Content-Type",
-            "Authorization",
-            HEADER_NAME_APP_CODE,
-            HEADER_NAME_CSRF_TOKEN,
-            HEADER_NAME_PASSPORT,
-        ],
+        allow_headers=list(AUTHENTICATED_HEADERS),
         methods=["GET", "PUT", "POST", "DELETE", "OPTIONS", "PATCH"],
         expose_headers=["X-Version", "X-Env"],
     )

api/extensions/ext_socketio.py

@@ -1,3 +0,0 @@
-import socketio
-
-sio = socketio.Server(async_mode="gevent", cors_allowed_origins="*")

api/factories/variable_factory.py

@@ -21,8 +21,6 @@ from core.variables.segments import (
     ObjectSegment,
     Segment,
     StringSegment,
-    VersionedMemorySegment,
-    VersionedMemoryValue,
 )
 from core.variables.types import SegmentType
 from core.variables.variables import (
@@ -41,7 +39,6 @@ from core.variables.variables import (
     SecretVariable,
     StringVariable,
     Variable,
-    VersionedMemoryVariable,
 )
 from core.workflow.constants import (
     CONVERSATION_VARIABLE_NODE_ID,
@@ -72,7 +69,6 @@ SEGMENT_TO_VARIABLE_MAP = {
     NoneSegment: NoneVariable,
     ObjectSegment: ObjectVariable,
     StringSegment: StringVariable,
-    VersionedMemorySegment: VersionedMemoryVariable
 }
 
 
@@ -197,7 +193,6 @@ _segment_factory: Mapping[SegmentType, type[Segment]] = {
     SegmentType.FILE: FileSegment,
     SegmentType.BOOLEAN: BooleanSegment,
     SegmentType.OBJECT: ObjectSegment,
-    SegmentType.VERSIONED_MEMORY: VersionedMemorySegment,
     # Array types
     SegmentType.ARRAY_ANY: ArrayAnySegment,
     SegmentType.ARRAY_STRING: ArrayStringSegment,
@@ -264,12 +259,6 @@ def build_segment_with_type(segment_type: SegmentType, value: Any) -> Segment:
         else:
             raise TypeMismatchError(f"Type mismatch: expected {segment_type}, but got empty list")
 
-    if segment_type == SegmentType.VERSIONED_MEMORY:
-        return VersionedMemorySegment(
-            value_type=segment_type,
-            value=VersionedMemoryValue.model_validate(value)
-        )
-
     inferred_type = SegmentType.infer_segment_type(value)
     # Type compatibility checking
     if inferred_type is None:

api/fields/online_user_fields.py

@@ -1,17 +0,0 @@
-from flask_restx import fields
-
-online_user_partial_fields = {
-    "user_id": fields.String,
-    "username": fields.String,
-    "avatar": fields.String,
-    "sid": fields.String,
-}
-
-workflow_online_users_fields = {
-    "workflow_id": fields.String,
-    "users": fields.List(fields.Nested(online_user_partial_fields)),
-}
-
-online_user_list_fields = {
-    "data": fields.List(fields.Nested(workflow_online_users_fields)),
-}

api/fields/workflow_comment_fields.py

@@ -1,96 +0,0 @@
-from flask_restx import fields
-
-from libs.helper import AvatarUrlField, TimestampField
-
-# basic account fields for comments
-account_fields = {
-    "id": fields.String,
-    "name": fields.String,
-    "email": fields.String,
-    "avatar_url": AvatarUrlField,
-}
-
-# Comment mention fields
-workflow_comment_mention_fields = {
-    "mentioned_user_id": fields.String,
-    "mentioned_user_account": fields.Nested(account_fields, allow_null=True),
-    "reply_id": fields.String,
-}
-
-# Comment reply fields
-workflow_comment_reply_fields = {
-    "id": fields.String,
-    "content": fields.String,
-    "created_by": fields.String,
-    "created_by_account": fields.Nested(account_fields, allow_null=True),
-    "created_at": TimestampField,
-}
-
-# Basic comment fields (for list views)
-workflow_comment_basic_fields = {
-    "id": fields.String,
-    "position_x": fields.Float,
-    "position_y": fields.Float,
-    "content": fields.String,
-    "created_by": fields.String,
-    "created_by_account": fields.Nested(account_fields, allow_null=True),
-    "created_at": TimestampField,
-    "updated_at": TimestampField,
-    "resolved": fields.Boolean,
-    "resolved_at": TimestampField,
-    "resolved_by": fields.String,
-    "resolved_by_account": fields.Nested(account_fields, allow_null=True),
-    "reply_count": fields.Integer,
-    "mention_count": fields.Integer,
-    "participants": fields.List(fields.Nested(account_fields)),
-}
-
-# Detailed comment fields (for single comment view)
-workflow_comment_detail_fields = {
-    "id": fields.String,
-    "position_x": fields.Float,
-    "position_y": fields.Float,
-    "content": fields.String,
-    "created_by": fields.String,
-    "created_by_account": fields.Nested(account_fields, allow_null=True),
-    "created_at": TimestampField,
-    "updated_at": TimestampField,
-    "resolved": fields.Boolean,
-    "resolved_at": TimestampField,
-    "resolved_by": fields.String,
-    "resolved_by_account": fields.Nested(account_fields, allow_null=True),
-    "replies": fields.List(fields.Nested(workflow_comment_reply_fields)),
-    "mentions": fields.List(fields.Nested(workflow_comment_mention_fields)),
-}
-
-# Comment creation response fields (simplified)
-workflow_comment_create_fields = {
-    "id": fields.String,
-    "created_at": TimestampField,
-}
-
-# Comment update response fields (simplified)
-workflow_comment_update_fields = {
-    "id": fields.String,
-    "updated_at": TimestampField,
-}
-
-# Comment resolve response fields
-workflow_comment_resolve_fields = {
-    "id": fields.String,
-    "resolved": fields.Boolean,
-    "resolved_at": TimestampField,
-    "resolved_by": fields.String,
-}
-
-# Reply creation response fields (simplified)
-workflow_comment_reply_create_fields = {
-    "id": fields.String,
-    "created_at": TimestampField,
-}
-
-# Reply update response fields
-workflow_comment_reply_update_fields = {
-    "id": fields.String,
-    "updated_at": TimestampField,
-}

api/libs/broadcast_channel/channel.py

@@ -0,0 +1,134 @@
+"""
+Broadcast channel for Pub/Sub messaging.
+"""
+
+import types
+from abc import abstractmethod
+from collections.abc import Iterator
+from contextlib import AbstractContextManager
+from typing import Protocol, Self
+
+
+class Subscription(AbstractContextManager["Subscription"], Protocol):
+    """A subscription to a topic that provides an iterator over received messages.
+    The subscription can be used as a context manager and will automatically
+    close when exiting the context.
+
+    Note: `Subscription` instances are not thread-safe. Each thread should create its own
+    subscription.
+    """
+
+    @abstractmethod
+    def __iter__(self) -> Iterator[bytes]:
+        """`__iter__` returns an iterator used to consume the message from this subscription.
+
+        If the caller did not enter the context, `__iter__` may lazily perform the setup before
+        yielding messages; otherwise `__enter__` handles it.”
+
+        If the subscription is closed, then the returned  iterator exits without
+        raising any error.
+        """
+        ...
+
+    @abstractmethod
+    def close(self) -> None:
+        """close closes the subscription, releases any resources associated with it."""
+        ...
+
+    def __enter__(self) -> Self:
+        """`__enter__` does the setup logic of the subscription (if any), and return itself."""
+        return self
+
+    def __exit__(
+        self,
+        exc_type: type[BaseException] | None,
+        exc_value: BaseException | None,
+        traceback: types.TracebackType | None,
+    ) -> bool | None:
+        self.close()
+        return None
+
+    @abstractmethod
+    def receive(self, timeout: float | None = 0.1) -> bytes | None:
+        """Receive the next message from the broadcast channel.
+
+        If `timeout` is specified, this method returns `None` if no message is
+        received within the given period. If `timeout` is `None`, the call blocks
+        until a message is received.
+
+        Calling receive with `timeout=None` is highly discouraged, as it is impossible to
+        cancel a blocking subscription.
+
+        :param timeout: timeout for receive message, in seconds.
+
+        Returns:
+            bytes: The received message as a byte string, or
+            None: If the timeout expires before a message is received.
+
+        Raises:
+            SubscriptionClosed: If the subscription has already been closed.
+        """
+        ...
+
+
+class Producer(Protocol):
+    """Producer is an interface for message publishing. It is already bound to a specific topic.
+
+    `Producer` implementations must be thread-safe and support concurrent use by multiple threads.
+    """
+
+    @abstractmethod
+    def publish(self, payload: bytes) -> None:
+        """Publish a message to the bounded topic."""
+        ...
+
+
+class Subscriber(Protocol):
+    """Subscriber is an interface for subscription creation. It is already bound to a specific topic.
+
+    `Subscriber` implementations must be thread-safe and support concurrent use by multiple threads.
+    """
+
+    @abstractmethod
+    def subscribe(self) -> Subscription:
+        pass
+
+
+class Topic(Producer, Subscriber, Protocol):
+    """A named channel for publishing and subscribing to messages.
+
+    Topics provide both read and write access. For restricted access,
+    use as_producer() for write-only view or as_subscriber() for read-only view.
+
+    `Topic` implementations must be thread-safe and support concurrent use by multiple threads.
+    """
+
+    @abstractmethod
+    def as_producer(self) -> Producer:
+        """as_producer creates a write-only view for this topic."""
+        ...
+
+    @abstractmethod
+    def as_subscriber(self) -> Subscriber:
+        """as_subscriber create a read-only view for this topic."""
+        ...
+
+
+class BroadcastChannel(Protocol):
+    """A broadcasting channel is a channel supporting broadcasting semantics.
+
+    Each channel is identified by a topic, different topics are isolated and do not affect each other.
+
+    There can be multiple subscriptions to a specific topic. When a publisher publishes a message to
+    a specific topic, all subscription should receive the published message.
+
+    There are no restriction for the persistence of messages. Once a subscription is created, it
+    should receive all subsequent messages published.
+
+    `BroadcastChannel` implementations must be thread-safe and support concurrent use by multiple threads.
+    """
+
+    @abstractmethod
+    def topic(self, topic: str) -> "Topic":
+        """topic returns a `Topic` instance for the given topic name."""
+        ...

api/libs/broadcast_channel/exc.py

@@ -0,0 +1,12 @@
+class BroadcastChannelError(Exception):
+    """`BroadcastChannelError` is the base class for all exceptions related
+    to `BroadcastChannel`."""
+
+    pass
+
+
+class SubscriptionClosedError(BroadcastChannelError):
+    """SubscriptionClosedError means that the subscription has been closed and
+    methods for consuming messages should not be called."""
+
+    pass

api/libs/broadcast_channel/redis/__init__.py

@@ -0,0 +1,3 @@
+from .channel import BroadcastChannel
+
+__all__ = ["BroadcastChannel"]

api/libs/broadcast_channel/redis/channel.py

@@ -0,0 +1,200 @@
+import logging
+import queue
+import threading
+import types
+from collections.abc import Generator, Iterator
+from typing import Self
+
+from libs.broadcast_channel.channel import Producer, Subscriber, Subscription
+from libs.broadcast_channel.exc import SubscriptionClosedError
+from redis import Redis
+from redis.client import PubSub
+
+_logger = logging.getLogger(__name__)
+
+
+class BroadcastChannel:
+    """
+    Redis Pub/Sub based broadcast channel implementation.
+
+    Provides "at most once" delivery semantics for messages published to channels.
+    Uses Redis PUBLISH/SUBSCRIBE commands for real-time message delivery.
+
+    The `redis_client` used to construct BroadcastChannel should have `decode_responses` set to `False`.
+    """
+
+    def __init__(
+        self,
+        redis_client: Redis,
+    ):
+        self._client = redis_client
+
+    def topic(self, topic: str) -> "Topic":
+        return Topic(self._client, topic)
+
+
+class Topic:
+    def __init__(self, redis_client: Redis, topic: str):
+        self._client = redis_client
+        self._topic = topic
+
+    def as_producer(self) -> Producer:
+        return self
+
+    def publish(self, payload: bytes) -> None:
+        self._client.publish(self._topic, payload)
+
+    def as_subscriber(self) -> Subscriber:
+        return self
+
+    def subscribe(self) -> Subscription:
+        return _RedisSubscription(
+            pubsub=self._client.pubsub(),
+            topic=self._topic,
+        )
+
+
+class _RedisSubscription(Subscription):
+    def __init__(
+        self,
+        pubsub: PubSub,
+        topic: str,
+    ):
+        # The _pubsub is None only if the subscription is closed.
+        self._pubsub: PubSub | None = pubsub
+        self._topic = topic
+        self._closed = threading.Event()
+        self._queue: queue.Queue[bytes] = queue.Queue(maxsize=1024)
+        self._dropped_count = 0
+        self._listener_thread: threading.Thread | None = None
+        self._start_lock = threading.Lock()
+        self._started = False
+
+    def _start_if_needed(self) -> None:
+        with self._start_lock:
+            if self._started:
+                return
+            if self._closed.is_set():
+                raise SubscriptionClosedError("The Redis subscription is closed")
+            if self._pubsub is None:
+                raise SubscriptionClosedError("The Redis subscription has been cleaned up")
+
+            self._pubsub.subscribe(self._topic)
+            _logger.debug("Subscribed to channel %s", self._topic)
+
+            self._listener_thread = threading.Thread(
+                target=self._listen,
+                name=f"redis-broadcast-{self._topic}",
+                daemon=True,
+            )
+            self._listener_thread.start()
+            self._started = True
+
+    def _listen(self) -> None:
+        pubsub = self._pubsub
+        assert pubsub is not None, "PubSub should not be None while starting listening."
+        while not self._closed.is_set():
+            raw_message = pubsub.get_message(ignore_subscribe_messages=True, timeout=0.1)
+
+            if raw_message is None:
+                continue
+
+            if raw_message.get("type") != "message":
+                continue
+
+            channel_field = raw_message.get("channel")
+            if isinstance(channel_field, bytes):
+                channel_name = channel_field.decode("utf-8")
+            elif isinstance(channel_field, str):
+                channel_name = channel_field
+            else:
+                channel_name = str(channel_field)
+
+            if channel_name != self._topic:
+                _logger.warning("Ignoring message from unexpected channel %s", channel_name)
+                continue
+
+            payload_bytes: bytes | None = raw_message.get("data")
+            if not isinstance(payload_bytes, bytes):
+                _logger.error("Received invalid data from channel %s, type=%s", self._topic, type(payload_bytes))
+                continue
+
+            self._enqueue_message(payload_bytes)
+
+        _logger.debug("Listener thread stopped for channel %s", self._topic)
+        pubsub.unsubscribe(self._topic)
+        pubsub.close()
+        _logger.debug("PubSub closed for topic %s", self._topic)
+        self._pubsub = None
+
+    def _enqueue_message(self, payload: bytes) -> None:
+        while not self._closed.is_set():
+            try:
+                self._queue.put_nowait(payload)
+                return
+            except queue.Full:
+                try:
+                    self._queue.get_nowait()
+                    self._dropped_count += 1
+                    _logger.debug(
+                        "Dropped message from Redis subscription, topic=%s, total_dropped=%d",
+                        self._topic,
+                        self._dropped_count,
+                    )
+                except queue.Empty:
+                    continue
+        return
+
+    def _message_iterator(self) -> Generator[bytes, None, None]:
+        while not self._closed.is_set():
+            try:
+                item = self._queue.get(timeout=0.1)
+            except queue.Empty:
+                continue
+
+            yield item
+
+    def __iter__(self) -> Iterator[bytes]:
+        if self._closed.is_set():
+            raise SubscriptionClosedError("The Redis subscription is closed")
+        self._start_if_needed()
+        return iter(self._message_iterator())
+
+    def receive(self, timeout: float | None = None) -> bytes | None:
+        if self._closed.is_set():
+            raise SubscriptionClosedError("The Redis subscription is closed")
+        self._start_if_needed()
+
+        try:
+            item = self._queue.get(timeout=timeout)
+        except queue.Empty:
+            return None
+
+        return item
+
+    def __enter__(self) -> Self:
+        self._start_if_needed()
+        return self
+
+    def __exit__(
+        self,
+        exc_type: type[BaseException] | None,
+        exc_value: BaseException | None,
+        traceback: types.TracebackType | None,
+    ) -> bool | None:
+        self.close()
+        return None
+
+    def close(self) -> None:
+        if self._closed.is_set():
+            return
+
+        self._closed.set()
+        # NOTE: PubSub is not thread-safe. More specifically, the `PubSub.close` method and the `PubSub.get_message`
+        # method should NOT be called concurrently.
+        #
+        # Due to the restriction above, the PubSub cleanup logic happens inside the consumer thread.
+        listener = self._listener_thread
+        if listener is not None:
+            listener.join(timeout=1.0)
+            self._listener_thread = None

api/libs/helper.py

@@ -177,6 +177,15 @@ def timezone(timezone_string):
     raise ValueError(error)
 
 
+def convert_datetime_to_date(field, target_timezone: str = ":tz"):
+    if dify_config.SQLALCHEMY_DATABASE_URI_SCHEME == "postgresql":
+        return f"DATE(DATE_TRUNC('day', {field} AT TIME ZONE 'UTC' AT TIME ZONE {target_timezone}))"
+    elif "mysql" in dify_config.SQLALCHEMY_DATABASE_URI_SCHEME:
+        return f"DATE(CONVERT_TZ({field}, 'UTC', {target_timezone}))"
+    else:
+        raise NotImplementedError(f"Unsupported database URI scheme: {dify_config.SQLALCHEMY_DATABASE_URI_SCHEME}")
+
+
 def generate_string(n):
     letters_digits = string.ascii_letters + string.digits
     result = ""

api/migrations/versions/00bacef91f18_rename_api_provider_description.py

@@ -8,6 +8,12 @@ Create Date: 2024-01-07 04:07:34.482983
 import sqlalchemy as sa
 from alembic import op
 
+import models.types
+
+
+def _is_pg(conn):
+    return conn.dialect.name == "postgresql"
+
 # revision identifiers, used by Alembic.
 revision = '00bacef91f18'
 down_revision = '8ec536f3c800'
@@ -17,17 +23,31 @@ depends_on = None
 
 def upgrade():
     # ### commands auto generated by Alembic - please adjust! ###
-    with op.batch_alter_table('tool_api_providers', schema=None) as batch_op:
-        batch_op.add_column(sa.Column('description', sa.Text(), nullable=False))
-        batch_op.drop_column('description_str')
+    conn = op.get_bind()
+    
+    if _is_pg(conn):
+        with op.batch_alter_table('tool_api_providers', schema=None) as batch_op:
+            batch_op.add_column(sa.Column('description', sa.Text(), nullable=False))
+            batch_op.drop_column('description_str')
+    else:
+        with op.batch_alter_table('tool_api_providers', schema=None) as batch_op:
+            batch_op.add_column(sa.Column('description', models.types.LongText(), nullable=False))
+            batch_op.drop_column('description_str')
 
     # ### end Alembic commands ###
 
 
 def downgrade():
     # ### commands auto generated by Alembic - please adjust! ###
-    with op.batch_alter_table('tool_api_providers', schema=None) as batch_op:
-        batch_op.add_column(sa.Column('description_str', sa.TEXT(), autoincrement=False, nullable=False))
-        batch_op.drop_column('description')
+    conn = op.get_bind()
+    
+    if _is_pg(conn):
+        with op.batch_alter_table('tool_api_providers', schema=None) as batch_op:
+            batch_op.add_column(sa.Column('description_str', sa.TEXT(), autoincrement=False, nullable=False))
+            batch_op.drop_column('description')
+    else:
+        with op.batch_alter_table('tool_api_providers', schema=None) as batch_op:
+            batch_op.add_column(sa.Column('description_str', models.types.LongText(), autoincrement=False, nullable=False))
+            batch_op.drop_column('description')
 
     # ### end Alembic commands ###

api/migrations/versions/04c602f5dc9b_update_appmodelconfig_and_add_table_.py

@@ -7,9 +7,14 @@ Create Date: 2024-06-12 07:49:07.666510
 """
 import sqlalchemy as sa
 from alembic import op
+from uuid import uuid4
 
 import models.types
 
+
+def _is_pg(conn):
+    return conn.dialect.name == "postgresql"
+
 # revision identifiers, used by Alembic.
 revision = '04c602f5dc9b'
 down_revision = '4ff534e1eb11'
@@ -19,15 +24,28 @@ depends_on = None
 
 def upgrade():
     # ### commands auto generated by Alembic - please adjust! ###
-    op.create_table('tracing_app_configs',
-    sa.Column('id', models.types.StringUUID(), server_default=sa.text('uuid_generate_v4()'), nullable=False),
-    sa.Column('app_id', models.types.StringUUID(), nullable=False),
-    sa.Column('tracing_provider', sa.String(length=255), nullable=True),
-    sa.Column('tracing_config', sa.JSON(), nullable=True),
-    sa.Column('created_at', sa.DateTime(), server_default=sa.text('now()'), nullable=False),
-    sa.Column('updated_at', sa.DateTime(), server_default=sa.text('now()'), nullable=False),
-    sa.PrimaryKeyConstraint('id', name='tracing_app_config_pkey')
-    )
+    conn = op.get_bind()
+    
+    if _is_pg(conn):
+        op.create_table('tracing_app_configs',
+        sa.Column('id', models.types.StringUUID(), server_default=sa.text('uuid_generate_v4()'), nullable=False),
+        sa.Column('app_id', models.types.StringUUID(), nullable=False),
+        sa.Column('tracing_provider', sa.String(length=255), nullable=True),
+        sa.Column('tracing_config', sa.JSON(), nullable=True), 
+        sa.Column('created_at', sa.DateTime(), server_default=sa.text('now()'), nullable=False),
+        sa.Column('updated_at', sa.DateTime(), server_default=sa.text('now()'), nullable=False),
+        sa.PrimaryKeyConstraint('id', name='tracing_app_config_pkey')
+        )
+    else:
+        op.create_table('tracing_app_configs',
+        sa.Column('id', models.types.StringUUID(), default=lambda: str(uuid4()), nullable=False),
+        sa.Column('app_id', models.types.StringUUID(), nullable=False),
+        sa.Column('tracing_provider', sa.String(length=255), nullable=True),
+        sa.Column('tracing_config', sa.JSON(), nullable=True),  
+        sa.Column('created_at', sa.DateTime(), server_default=sa.func.now(), nullable=False),
+        sa.Column('updated_at', sa.DateTime(), server_default=sa.func.now(), nullable=False),
+        sa.PrimaryKeyConstraint('id', name='tracing_app_config_pkey')
+        )
 
     # ### end Alembic commands ###
 

api/migrations/versions/053da0c1d756_add_api_tool_privacy.py

@@ -7,8 +7,15 @@ Create Date: 2024-01-12 06:47:21.656262
 """
 import sqlalchemy as sa
 from alembic import op
+from uuid import uuid4
 from sqlalchemy.dialects import postgresql
 
+import models.types
+
+
+def _is_pg(conn):
+    return conn.dialect.name == "postgresql"
+
 # revision identifiers, used by Alembic.
 revision = '053da0c1d756'
 down_revision = '4829e54d2fee'
@@ -18,16 +25,31 @@ depends_on = None
 
 def upgrade():
     # ### commands auto generated by Alembic - please adjust! ###
-    op.create_table('tool_conversation_variables',
-    sa.Column('id', postgresql.UUID(), server_default=sa.text('uuid_generate_v4()'), nullable=False),
-    sa.Column('user_id', postgresql.UUID(), nullable=False),
-    sa.Column('tenant_id', postgresql.UUID(), nullable=False),
-    sa.Column('conversation_id', postgresql.UUID(), nullable=False),
-    sa.Column('variables_str', sa.Text(), nullable=False),
-    sa.Column('created_at', sa.DateTime(), server_default=sa.text('CURRENT_TIMESTAMP(0)'), nullable=False),
-    sa.Column('updated_at', sa.DateTime(), server_default=sa.text('CURRENT_TIMESTAMP(0)'), nullable=False),
-    sa.PrimaryKeyConstraint('id', name='tool_conversation_variables_pkey')
-    )
+    conn = op.get_bind()
+    
+    if _is_pg(conn):
+        op.create_table('tool_conversation_variables',
+        sa.Column('id', postgresql.UUID(), server_default=sa.text('uuid_generate_v4()'), nullable=False),
+        sa.Column('user_id', postgresql.UUID(), nullable=False),
+        sa.Column('tenant_id', postgresql.UUID(), nullable=False),
+        sa.Column('conversation_id', postgresql.UUID(), nullable=False),
+        sa.Column('variables_str', sa.Text(), nullable=False),
+        sa.Column('created_at', sa.DateTime(), server_default=sa.text('CURRENT_TIMESTAMP(0)'), nullable=False),
+        sa.Column('updated_at', sa.DateTime(), server_default=sa.text('CURRENT_TIMESTAMP(0)'), nullable=False),
+        sa.PrimaryKeyConstraint('id', name='tool_conversation_variables_pkey')
+        )
+    else:
+        op.create_table('tool_conversation_variables',
+        sa.Column('id', models.types.StringUUID(), default=lambda: str(uuid4()), nullable=False),
+        sa.Column('user_id', models.types.StringUUID(), nullable=False),
+        sa.Column('tenant_id', models.types.StringUUID(), nullable=False),
+        sa.Column('conversation_id', models.types.StringUUID(), nullable=False),
+        sa.Column('variables_str', models.types.LongText(), nullable=False),
+        sa.Column('created_at', sa.DateTime(), server_default=sa.func.current_timestamp(), nullable=False),
+        sa.Column('updated_at', sa.DateTime(), server_default=sa.func.current_timestamp(), nullable=False),
+        sa.PrimaryKeyConstraint('id', name='tool_conversation_variables_pkey')
+        )
+    
     with op.batch_alter_table('tool_api_providers', schema=None) as batch_op:
         batch_op.add_column(sa.Column('privacy_policy', sa.String(length=255), nullable=True))
         batch_op.alter_column('icon',

api/migrations/versions/114eed84c228_remove_tool_id_from_model_invoke.py

@@ -9,6 +9,12 @@ import sqlalchemy as sa
 from alembic import op
 from sqlalchemy.dialects import postgresql
 
+import models.types
+
+
+def _is_pg(conn):
+    return conn.dialect.name == "postgresql"
+
 # revision identifiers, used by Alembic.
 revision = '114eed84c228'
 down_revision = 'c71211c8f604'
@@ -26,7 +32,13 @@ def upgrade():
 
 def downgrade():
     # ### commands auto generated by Alembic - please adjust! ###
-    with op.batch_alter_table('tool_model_invokes', schema=None) as batch_op:
-        batch_op.add_column(sa.Column('tool_id', postgresql.UUID(), autoincrement=False, nullable=False))
+    conn = op.get_bind()
+    
+    if _is_pg(conn):
+        with op.batch_alter_table('tool_model_invokes', schema=None) as batch_op:
+            batch_op.add_column(sa.Column('tool_id', postgresql.UUID(), autoincrement=False, nullable=False))
+    else:
+        with op.batch_alter_table('tool_model_invokes', schema=None) as batch_op:
+            batch_op.add_column(sa.Column('tool_id', models.types.StringUUID(), autoincrement=False, nullable=False))
 
     # ### end Alembic commands ###

api/migrations/versions/161cadc1af8d_add_dataset_permission_tenant_id.py

@@ -8,7 +8,11 @@ Create Date: 2024-07-05 14:30:59.472593
 import sqlalchemy as sa
 from alembic import op
 
-import models as models
+import models.types
+
+
+def _is_pg(conn):
+    return conn.dialect.name == "postgresql"
 
 # revision identifiers, used by Alembic.
 revision = '161cadc1af8d'
@@ -19,9 +23,16 @@ depends_on = None
 
 def upgrade():
     # ### commands auto generated by Alembic - please adjust! ###
-    with op.batch_alter_table('dataset_permissions', schema=None) as batch_op:
-        # Step 1: Add column without NOT NULL constraint
-        op.add_column('dataset_permissions', sa.Column('tenant_id', sa.UUID(), nullable=False))
+    conn = op.get_bind()
+    
+    if _is_pg(conn):
+        with op.batch_alter_table('dataset_permissions', schema=None) as batch_op:
+            # Step 1: Add column without NOT NULL constraint
+            op.add_column('dataset_permissions', sa.Column('tenant_id', sa.UUID(), nullable=False))
+    else:
+        with op.batch_alter_table('dataset_permissions', schema=None) as batch_op:
+            # Step 1: Add column without NOT NULL constraint
+            op.add_column('dataset_permissions', sa.Column('tenant_id', models.types.StringUUID(), nullable=False))
 
     # ### end Alembic commands ###
 

api/migrations/versions/16fa53d9faec_add_provider_model_support.py

@@ -7,8 +7,15 @@ Create Date: 2023-08-06 16:57:51.248337
 """
 import sqlalchemy as sa
 from alembic import op
+from uuid import uuid4
 from sqlalchemy.dialects import postgresql
 
+import models.types
+
+
+def _is_pg(conn):
+    return conn.dialect.name == "postgresql"
+
 # revision identifiers, used by Alembic.
 revision = '16fa53d9faec'
 down_revision = '8d2d099ceb74'
@@ -18,44 +25,87 @@ depends_on = None
 
 def upgrade():
     # ### commands auto generated by Alembic - please adjust! ###
-    op.create_table('provider_models',
-    sa.Column('id', postgresql.UUID(), server_default=sa.text('uuid_generate_v4()'), nullable=False),
-    sa.Column('tenant_id', postgresql.UUID(), nullable=False),
-    sa.Column('provider_name', sa.String(length=40), nullable=False),
-    sa.Column('model_name', sa.String(length=40), nullable=False),
-    sa.Column('model_type', sa.String(length=40), nullable=False),
-    sa.Column('encrypted_config', sa.Text(), nullable=True),
-    sa.Column('is_valid', sa.Boolean(), server_default=sa.text('false'), nullable=False),
-    sa.Column('created_at', sa.DateTime(), server_default=sa.text('CURRENT_TIMESTAMP(0)'), nullable=False),
-    sa.Column('updated_at', sa.DateTime(), server_default=sa.text('CURRENT_TIMESTAMP(0)'), nullable=False),
-    sa.PrimaryKeyConstraint('id', name='provider_model_pkey'),
-    sa.UniqueConstraint('tenant_id', 'provider_name', 'model_name', 'model_type', name='unique_provider_model_name')
-    )
+    conn = op.get_bind()
+    
+    if _is_pg(conn):
+        op.create_table('provider_models',
+        sa.Column('id', postgresql.UUID(), server_default=sa.text('uuid_generate_v4()'), nullable=False),
+        sa.Column('tenant_id', postgresql.UUID(), nullable=False),
+        sa.Column('provider_name', sa.String(length=40), nullable=False),
+        sa.Column('model_name', sa.String(length=40), nullable=False),
+        sa.Column('model_type', sa.String(length=40), nullable=False),
+        sa.Column('encrypted_config', sa.Text(), nullable=True),
+        sa.Column('is_valid', sa.Boolean(), server_default=sa.text('false'), nullable=False),
+        sa.Column('created_at', sa.DateTime(), server_default=sa.text('CURRENT_TIMESTAMP(0)'), nullable=False),
+        sa.Column('updated_at', sa.DateTime(), server_default=sa.text('CURRENT_TIMESTAMP(0)'), nullable=False),
+        sa.PrimaryKeyConstraint('id', name='provider_model_pkey'),
+        sa.UniqueConstraint('tenant_id', 'provider_name', 'model_name', 'model_type', name='unique_provider_model_name')
+        )
+    else:
+        op.create_table('provider_models',
+        sa.Column('id', models.types.StringUUID(), default=lambda: str(uuid4()), nullable=False),
+        sa.Column('tenant_id', models.types.StringUUID(), nullable=False),
+        sa.Column('provider_name', sa.String(length=40), nullable=False),
+        sa.Column('model_name', sa.String(length=40), nullable=False),
+        sa.Column('model_type', sa.String(length=40), nullable=False),
+        sa.Column('encrypted_config', models.types.LongText(), nullable=True),
+        sa.Column('is_valid', sa.Boolean(), server_default=sa.text('false'), nullable=False),
+        sa.Column('created_at', sa.DateTime(), server_default=sa.func.current_timestamp(), nullable=False),
+        sa.Column('updated_at', sa.DateTime(), server_default=sa.func.current_timestamp(), nullable=False),
+        sa.PrimaryKeyConstraint('id', name='provider_model_pkey'),
+        sa.UniqueConstraint('tenant_id', 'provider_name', 'model_name', 'model_type', name='unique_provider_model_name')
+        )
+    
     with op.batch_alter_table('provider_models', schema=None) as batch_op:
         batch_op.create_index('provider_model_tenant_id_provider_idx', ['tenant_id', 'provider_name'], unique=False)
 
-    op.create_table('tenant_default_models',
-    sa.Column('id', postgresql.UUID(), server_default=sa.text('uuid_generate_v4()'), nullable=False),
-    sa.Column('tenant_id', postgresql.UUID(), nullable=False),
-    sa.Column('provider_name', sa.String(length=40), nullable=False),
-    sa.Column('model_name', sa.String(length=40), nullable=False),
-    sa.Column('model_type', sa.String(length=40), nullable=False),
-    sa.Column('created_at', sa.DateTime(), server_default=sa.text('CURRENT_TIMESTAMP(0)'), nullable=False),
-    sa.Column('updated_at', sa.DateTime(), server_default=sa.text('CURRENT_TIMESTAMP(0)'), nullable=False),
-    sa.PrimaryKeyConstraint('id', name='tenant_default_model_pkey')
-    )
+    if _is_pg(conn):
+        op.create_table('tenant_default_models',
+        sa.Column('id', postgresql.UUID(), server_default=sa.text('uuid_generate_v4()'), nullable=False),
+        sa.Column('tenant_id', postgresql.UUID(), nullable=False),
+        sa.Column('provider_name', sa.String(length=40), nullable=False),
+        sa.Column('model_name', sa.String(length=40), nullable=False),
+        sa.Column('model_type', sa.String(length=40), nullable=False),
+        sa.Column('created_at', sa.DateTime(), server_default=sa.text('CURRENT_TIMESTAMP(0)'), nullable=False),
+        sa.Column('updated_at', sa.DateTime(), server_default=sa.text('CURRENT_TIMESTAMP(0)'), nullable=False),
+        sa.PrimaryKeyConstraint('id', name='tenant_default_model_pkey')
+        )
+    else:
+        op.create_table('tenant_default_models',
+        sa.Column('id', models.types.StringUUID(), default=lambda: str(uuid4()), nullable=False),
+        sa.Column('tenant_id', models.types.StringUUID(), nullable=False),
+        sa.Column('provider_name', sa.String(length=40), nullable=False),
+        sa.Column('model_name', sa.String(length=40), nullable=False),
+        sa.Column('model_type', sa.String(length=40), nullable=False),
+        sa.Column('created_at', sa.DateTime(), server_default=sa.func.current_timestamp(), nullable=False),
+        sa.Column('updated_at', sa.DateTime(), server_default=sa.func.current_timestamp(), nullable=False),
+        sa.PrimaryKeyConstraint('id', name='tenant_default_model_pkey')
+        )
+    
     with op.batch_alter_table('tenant_default_models', schema=None) as batch_op:
         batch_op.create_index('tenant_default_model_tenant_id_provider_type_idx', ['tenant_id', 'provider_name', 'model_type'], unique=False)
 
-    op.create_table('tenant_preferred_model_providers',
-    sa.Column('id', postgresql.UUID(), server_default=sa.text('uuid_generate_v4()'), nullable=False),
-    sa.Column('tenant_id', postgresql.UUID(), nullable=False),
-    sa.Column('provider_name', sa.String(length=40), nullable=False),
-    sa.Column('preferred_provider_type', sa.String(length=40), nullable=False),
-    sa.Column('created_at', sa.DateTime(), server_default=sa.text('CURRENT_TIMESTAMP(0)'), nullable=False),
-    sa.Column('updated_at', sa.DateTime(), server_default=sa.text('CURRENT_TIMESTAMP(0)'), nullable=False),
-    sa.PrimaryKeyConstraint('id', name='tenant_preferred_model_provider_pkey')
-    )
+    if _is_pg(conn):
+        op.create_table('tenant_preferred_model_providers',
+        sa.Column('id', postgresql.UUID(), server_default=sa.text('uuid_generate_v4()'), nullable=False),
+        sa.Column('tenant_id', postgresql.UUID(), nullable=False),
+        sa.Column('provider_name', sa.String(length=40), nullable=False),
+        sa.Column('preferred_provider_type', sa.String(length=40), nullable=False),
+        sa.Column('created_at', sa.DateTime(), server_default=sa.text('CURRENT_TIMESTAMP(0)'), nullable=False),
+        sa.Column('updated_at', sa.DateTime(), server_default=sa.text('CURRENT_TIMESTAMP(0)'), nullable=False),
+        sa.PrimaryKeyConstraint('id', name='tenant_preferred_model_provider_pkey')
+        )
+    else:
+        op.create_table('tenant_preferred_model_providers',
+        sa.Column('id', models.types.StringUUID(), default=lambda: str(uuid4()), nullable=False),
+        sa.Column('tenant_id', models.types.StringUUID(), nullable=False),
+        sa.Column('provider_name', sa.String(length=40), nullable=False),
+        sa.Column('preferred_provider_type', sa.String(length=40), nullable=False),
+        sa.Column('created_at', sa.DateTime(), server_default=sa.func.current_timestamp(), nullable=False),
+        sa.Column('updated_at', sa.DateTime(), server_default=sa.func.current_timestamp(), nullable=False),
+        sa.PrimaryKeyConstraint('id', name='tenant_preferred_model_provider_pkey')
+        )
+    
     with op.batch_alter_table('tenant_preferred_model_providers', schema=None) as batch_op:
         batch_op.create_index('tenant_preferred_model_provider_tenant_provider_idx', ['tenant_id', 'provider_name'], unique=False)
 

api/migrations/versions/17b5ab037c40_add_keyworg_table_storage_type.py

@@ -8,6 +8,10 @@ Create Date: 2024-04-01 09:48:54.232201
 import sqlalchemy as sa
 from alembic import op
 
+
+def _is_pg(conn):
+    return conn.dialect.name == "postgresql"
+
 # revision identifiers, used by Alembic.
 revision = '17b5ab037c40'
 down_revision = 'a8f9b3c45e4a'
@@ -17,9 +21,14 @@ depends_on = None
 
 def upgrade():
     # ### commands auto generated by Alembic - please adjust! ###
-
-    with op.batch_alter_table('dataset_keyword_tables', schema=None) as batch_op:
-        batch_op.add_column(sa.Column('data_source_type', sa.String(length=255), server_default=sa.text("'database'::character varying"), nullable=False))
+    conn = op.get_bind()
+    
+    if _is_pg(conn):
+        with op.batch_alter_table('dataset_keyword_tables', schema=None) as batch_op:
+            batch_op.add_column(sa.Column('data_source_type', sa.String(length=255), server_default=sa.text("'database'::character varying"), nullable=False))
+    else:
+        with op.batch_alter_table('dataset_keyword_tables', schema=None) as batch_op:
+            batch_op.add_column(sa.Column('data_source_type', sa.String(length=255), server_default=sa.text("'database'"), nullable=False))
 
     # ### end Alembic commands ###
 

api/migrations/versions/2024_08_13_0633-63a83fcf12ba_support_conversation_variables.py

@@ -10,6 +10,10 @@ from alembic import op
 
 import models as models
 
+
+def _is_pg(conn):
+    return conn.dialect.name == "postgresql"
+
 # revision identifiers, used by Alembic.
 revision = '63a83fcf12ba'
 down_revision = '1787fbae959a'
@@ -19,21 +23,39 @@ depends_on = None
 
 def upgrade():
     # ### commands auto generated by Alembic - please adjust! ###
-    op.create_table('workflow__conversation_variables',
-    sa.Column('id', models.types.StringUUID(), nullable=False),
-    sa.Column('conversation_id', models.types.StringUUID(), nullable=False),
-    sa.Column('app_id', models.types.StringUUID(), nullable=False),
-    sa.Column('data', sa.Text(), nullable=False),
-    sa.Column('created_at', sa.DateTime(), server_default=sa.text('CURRENT_TIMESTAMP(0)'), nullable=False),
-    sa.Column('updated_at', sa.DateTime(), server_default=sa.text('CURRENT_TIMESTAMP'), nullable=False),
-    sa.PrimaryKeyConstraint('id', 'conversation_id', name=op.f('workflow__conversation_variables_pkey'))
-    )
+    conn = op.get_bind()
+    
+    if _is_pg(conn):
+        op.create_table('workflow__conversation_variables',
+        sa.Column('id', models.types.StringUUID(), nullable=False),
+        sa.Column('conversation_id', models.types.StringUUID(), nullable=False),
+        sa.Column('app_id', models.types.StringUUID(), nullable=False),
+        sa.Column('data', sa.Text(), nullable=False),
+        sa.Column('created_at', sa.DateTime(), server_default=sa.text('CURRENT_TIMESTAMP(0)'), nullable=False),
+        sa.Column('updated_at', sa.DateTime(), server_default=sa.text('CURRENT_TIMESTAMP'), nullable=False),
+        sa.PrimaryKeyConstraint('id', 'conversation_id', name=op.f('workflow__conversation_variables_pkey'))
+        )
+    else:
+        op.create_table('workflow__conversation_variables',
+        sa.Column('id', models.types.StringUUID(), nullable=False),
+        sa.Column('conversation_id', models.types.StringUUID(), nullable=False),
+        sa.Column('app_id', models.types.StringUUID(), nullable=False),
+        sa.Column('data', models.types.LongText(), nullable=False),
+        sa.Column('created_at', sa.DateTime(), server_default=sa.func.current_timestamp(), nullable=False),
+        sa.Column('updated_at', sa.DateTime(), server_default=sa.func.current_timestamp(), nullable=False),
+        sa.PrimaryKeyConstraint('id', 'conversation_id', name=op.f('workflow__conversation_variables_pkey'))
+        )
+    
     with op.batch_alter_table('workflow__conversation_variables', schema=None) as batch_op:
         batch_op.create_index(batch_op.f('workflow__conversation_variables_app_id_idx'), ['app_id'], unique=False)
         batch_op.create_index(batch_op.f('workflow__conversation_variables_created_at_idx'), ['created_at'], unique=False)
 
-    with op.batch_alter_table('workflows', schema=None) as batch_op:
-        batch_op.add_column(sa.Column('conversation_variables', sa.Text(), server_default='{}', nullable=False))
+    if _is_pg(conn):
+        with op.batch_alter_table('workflows', schema=None) as batch_op:
+            batch_op.add_column(sa.Column('conversation_variables', sa.Text(), server_default='{}', nullable=False))
+    else:
+        with op.batch_alter_table('workflows', schema=None) as batch_op:
+            batch_op.add_column(sa.Column('conversation_variables', models.types.LongText(), default='{}', nullable=False))
 
     # ### end Alembic commands ###
 

api/migrations/versions/2024_08_15_0956-0251a1c768cc_add_tidb_auth_binding.py

@@ -7,9 +7,14 @@ Create Date: 2024-08-15 09:56:59.012490
 """
 import sqlalchemy as sa
 from alembic import op
+from uuid import uuid4
 
 import models as models
 
+
+def _is_pg(conn):
+    return conn.dialect.name == "postgresql"
+
 # revision identifiers, used by Alembic.
 revision = '0251a1c768cc'
 down_revision = 'bbadea11becb'
@@ -19,18 +24,35 @@ depends_on = None
 
 def upgrade():
     # ### commands auto generated by Alembic - please adjust! ###
-    op.create_table('tidb_auth_bindings',
-    sa.Column('id', models.types.StringUUID(), server_default=sa.text('uuid_generate_v4()'), nullable=False),
-    sa.Column('tenant_id', models.types.StringUUID(), nullable=True),
-    sa.Column('cluster_id', sa.String(length=255), nullable=False),
-    sa.Column('cluster_name', sa.String(length=255), nullable=False),
-    sa.Column('active', sa.Boolean(), server_default=sa.text('false'), nullable=False),
-    sa.Column('status', sa.String(length=255), server_default=sa.text("'CREATING'::character varying"), nullable=False),
-    sa.Column('account', sa.String(length=255), nullable=False),
-    sa.Column('password', sa.String(length=255), nullable=False),
-    sa.Column('created_at', sa.DateTime(), server_default=sa.text('CURRENT_TIMESTAMP(0)'), nullable=False),
-    sa.PrimaryKeyConstraint('id', name='tidb_auth_bindings_pkey')
-    )
+    conn = op.get_bind()
+    
+    if _is_pg(conn):
+        op.create_table('tidb_auth_bindings',
+        sa.Column('id', models.types.StringUUID(), server_default=sa.text('uuid_generate_v4()'), nullable=False),
+        sa.Column('tenant_id', models.types.StringUUID(), nullable=True),
+        sa.Column('cluster_id', sa.String(length=255), nullable=False),
+        sa.Column('cluster_name', sa.String(length=255), nullable=False),
+        sa.Column('active', sa.Boolean(), server_default=sa.text('false'), nullable=False),
+        sa.Column('status', sa.String(length=255), server_default=sa.text("'CREATING'::character varying"), nullable=False),
+        sa.Column('account', sa.String(length=255), nullable=False),
+        sa.Column('password', sa.String(length=255), nullable=False),
+        sa.Column('created_at', sa.DateTime(), server_default=sa.text('CURRENT_TIMESTAMP(0)'), nullable=False),
+        sa.PrimaryKeyConstraint('id', name='tidb_auth_bindings_pkey')
+        )
+    else:
+        op.create_table('tidb_auth_bindings',
+        sa.Column('id', models.types.StringUUID(), default=lambda: str(uuid4()), nullable=False),
+        sa.Column('tenant_id', models.types.StringUUID(), nullable=True),
+        sa.Column('cluster_id', sa.String(length=255), nullable=False),
+        sa.Column('cluster_name', sa.String(length=255), nullable=False),
+        sa.Column('active', sa.Boolean(), server_default=sa.text('false'), nullable=False),
+        sa.Column('status', sa.String(length=255), server_default=sa.text("'CREATING'"), nullable=False),
+        sa.Column('account', sa.String(length=255), nullable=False),
+        sa.Column('password', sa.String(length=255), nullable=False),
+        sa.Column('created_at', sa.DateTime(), server_default=sa.func.current_timestamp(), nullable=False),
+        sa.PrimaryKeyConstraint('id', name='tidb_auth_bindings_pkey')
+        )
+    
     with op.batch_alter_table('tidb_auth_bindings', schema=None) as batch_op:
         batch_op.create_index('tidb_auth_bindings_active_idx', ['active'], unique=False)
         batch_op.create_index('tidb_auth_bindings_status_idx', ['status'], unique=False)

api/migrations/versions/2024_09_11_1012-d57ba9ebb251_add_parent_message_id_to_messages.py

@@ -10,6 +10,10 @@ from alembic import op
 
 import models as models
 
+
+def _is_pg(conn):
+    return conn.dialect.name == "postgresql"
+
 # revision identifiers, used by Alembic.
 revision = 'd57ba9ebb251'
 down_revision = '675b5321501b'
@@ -22,8 +26,14 @@ def upgrade():
     with op.batch_alter_table('messages', schema=None) as batch_op:
         batch_op.add_column(sa.Column('parent_message_id', models.types.StringUUID(), nullable=True))
 
-    # Set parent_message_id for existing messages to uuid_nil() to distinguish them from new messages with actual parent IDs or NULLs
-    op.execute('UPDATE messages SET parent_message_id = uuid_nil() WHERE parent_message_id IS NULL')
+    # Set parent_message_id for existing messages to distinguish them from new messages with actual parent IDs or NULLs
+    conn = op.get_bind()
+    if _is_pg(conn):
+        # PostgreSQL: Use uuid_nil() function
+        op.execute('UPDATE messages SET parent_message_id = uuid_nil() WHERE parent_message_id IS NULL')
+    else:
+        # MySQL: Use a specific UUID value to represent nil
+        op.execute("UPDATE messages SET parent_message_id = '00000000-0000-0000-0000-000000000000' WHERE parent_message_id IS NULL")
 
     # ### end Alembic commands ###
 

api/migrations/versions/2024_09_24_0922-6af6a521a53e_update_retrieval_resource.py

@@ -6,7 +6,11 @@ Create Date: 2024-09-24 09:22:43.570120
 
 """
 from alembic import op
-import models as models
+import models.types
+
+
+def _is_pg(conn):
+    return conn.dialect.name == "postgresql"
 import sqlalchemy as sa
 from sqlalchemy.dialects import postgresql
 
@@ -19,30 +23,58 @@ depends_on = None
 
 def upgrade():
     # ### commands auto generated by Alembic - please adjust! ###
-    with op.batch_alter_table('dataset_retriever_resources', schema=None) as batch_op:
-        batch_op.alter_column('document_id',
-               existing_type=sa.UUID(),
-               nullable=True)
-        batch_op.alter_column('data_source_type',
-               existing_type=sa.TEXT(),
-               nullable=True)
-        batch_op.alter_column('segment_id',
-               existing_type=sa.UUID(),
-               nullable=True)
+    conn = op.get_bind()
+    
+    if _is_pg(conn):
+        with op.batch_alter_table('dataset_retriever_resources', schema=None) as batch_op:
+            batch_op.alter_column('document_id',
+                   existing_type=sa.UUID(),
+                   nullable=True)
+            batch_op.alter_column('data_source_type',
+                   existing_type=sa.TEXT(),
+                   nullable=True)
+            batch_op.alter_column('segment_id',
+                   existing_type=sa.UUID(),
+                   nullable=True)
+    else:
+        with op.batch_alter_table('dataset_retriever_resources', schema=None) as batch_op:
+            batch_op.alter_column('document_id',
+                   existing_type=models.types.StringUUID(),
+                   nullable=True)
+            batch_op.alter_column('data_source_type',
+                   existing_type=models.types.LongText(),
+                   nullable=True)
+            batch_op.alter_column('segment_id',
+                   existing_type=models.types.StringUUID(),
+                   nullable=True)
     # ### end Alembic commands ###
 
 
 def downgrade():
     # ### commands auto generated by Alembic - please adjust! ###
-    with op.batch_alter_table('dataset_retriever_resources', schema=None) as batch_op:
-        batch_op.alter_column('segment_id',
-               existing_type=sa.UUID(),
-               nullable=False)
-        batch_op.alter_column('data_source_type',
-               existing_type=sa.TEXT(),
-               nullable=False)
-        batch_op.alter_column('document_id',
-               existing_type=sa.UUID(),
-               nullable=False)
+    conn = op.get_bind()
+    
+    if _is_pg(conn):
+        with op.batch_alter_table('dataset_retriever_resources', schema=None) as batch_op:
+            batch_op.alter_column('segment_id',
+                   existing_type=sa.UUID(),
+                   nullable=False)
+            batch_op.alter_column('data_source_type',
+                   existing_type=sa.TEXT(),
+                   nullable=False)
+            batch_op.alter_column('document_id',
+                   existing_type=sa.UUID(),
+                   nullable=False)
+    else:
+        with op.batch_alter_table('dataset_retriever_resources', schema=None) as batch_op:
+            batch_op.alter_column('segment_id',
+                   existing_type=models.types.StringUUID(),
+                   nullable=False)
+            batch_op.alter_column('data_source_type',
+                   existing_type=models.types.LongText(),
+                   nullable=False)
+            batch_op.alter_column('document_id',
+                   existing_type=models.types.StringUUID(),
+                   nullable=False)
 
     # ### end Alembic commands ###

api/migrations/versions/2024_09_25_0434-33f5fac87f29_external_knowledge_api.py

@@ -9,6 +9,11 @@ from alembic import op
 import models as models
 import sqlalchemy as sa
 from sqlalchemy.dialects import postgresql
+from uuid import uuid4
+
+
+def _is_pg(conn):
+    return conn.dialect.name == "postgresql"
 
 # revision identifiers, used by Alembic.
 revision = '33f5fac87f29'
@@ -19,34 +24,66 @@ depends_on = None
 
 def upgrade():
     # ### commands auto generated by Alembic - please adjust! ###
-    op.create_table('external_knowledge_apis',
-    sa.Column('id', models.types.StringUUID(), server_default=sa.text('uuid_generate_v4()'), nullable=False),
-    sa.Column('name', sa.String(length=255), nullable=False),
-    sa.Column('description', sa.String(length=255), nullable=False),
-    sa.Column('tenant_id', models.types.StringUUID(), nullable=False),
-    sa.Column('settings', sa.Text(), nullable=True),
-    sa.Column('created_by', models.types.StringUUID(), nullable=False),
-    sa.Column('created_at', sa.DateTime(), server_default=sa.text('CURRENT_TIMESTAMP(0)'), nullable=False),
-    sa.Column('updated_by', models.types.StringUUID(), nullable=True),
-    sa.Column('updated_at', sa.DateTime(), server_default=sa.text('CURRENT_TIMESTAMP(0)'), nullable=False),
-    sa.PrimaryKeyConstraint('id', name='external_knowledge_apis_pkey')
-    )
+    conn = op.get_bind()
+    
+    if _is_pg(conn):
+        op.create_table('external_knowledge_apis',
+        sa.Column('id', models.types.StringUUID(), server_default=sa.text('uuid_generate_v4()'), nullable=False),
+        sa.Column('name', sa.String(length=255), nullable=False),
+        sa.Column('description', sa.String(length=255), nullable=False),
+        sa.Column('tenant_id', models.types.StringUUID(), nullable=False),
+        sa.Column('settings', sa.Text(), nullable=True),
+        sa.Column('created_by', models.types.StringUUID(), nullable=False),
+        sa.Column('created_at', sa.DateTime(), server_default=sa.text('CURRENT_TIMESTAMP(0)'), nullable=False),
+        sa.Column('updated_by', models.types.StringUUID(), nullable=True),
+        sa.Column('updated_at', sa.DateTime(), server_default=sa.text('CURRENT_TIMESTAMP(0)'), nullable=False),
+        sa.PrimaryKeyConstraint('id', name='external_knowledge_apis_pkey')
+        )
+    else:
+        op.create_table('external_knowledge_apis',
+        sa.Column('id', models.types.StringUUID(), default=lambda: str(uuid4()), nullable=False),
+        sa.Column('name', sa.String(length=255), nullable=False),
+        sa.Column('description', sa.String(length=255), nullable=False),
+        sa.Column('tenant_id', models.types.StringUUID(), nullable=False),
+        sa.Column('settings', models.types.LongText(), nullable=True),
+        sa.Column('created_by', models.types.StringUUID(), nullable=False),
+        sa.Column('created_at', sa.DateTime(), server_default=sa.func.current_timestamp(), nullable=False),
+        sa.Column('updated_by', models.types.StringUUID(), nullable=True),
+        sa.Column('updated_at', sa.DateTime(), server_default=sa.func.current_timestamp(), nullable=False),
+        sa.PrimaryKeyConstraint('id', name='external_knowledge_apis_pkey')
+        )
+    
     with op.batch_alter_table('external_knowledge_apis', schema=None) as batch_op:
         batch_op.create_index('external_knowledge_apis_name_idx', ['name'], unique=False)
         batch_op.create_index('external_knowledge_apis_tenant_idx', ['tenant_id'], unique=False)
 
-    op.create_table('external_knowledge_bindings',
-    sa.Column('id', models.types.StringUUID(), server_default=sa.text('uuid_generate_v4()'), nullable=False),
-    sa.Column('tenant_id', models.types.StringUUID(), nullable=False),
-    sa.Column('external_knowledge_api_id', models.types.StringUUID(), nullable=False),
-    sa.Column('dataset_id', models.types.StringUUID(), nullable=False),
-    sa.Column('external_knowledge_id', sa.Text(), nullable=False),
-    sa.Column('created_by', models.types.StringUUID(), nullable=False),
-    sa.Column('created_at', sa.DateTime(), server_default=sa.text('CURRENT_TIMESTAMP(0)'), nullable=False),
-    sa.Column('updated_by', models.types.StringUUID(), nullable=True),
-    sa.Column('updated_at', sa.DateTime(), server_default=sa.text('CURRENT_TIMESTAMP(0)'), nullable=False),
-    sa.PrimaryKeyConstraint('id', name='external_knowledge_bindings_pkey')
-    )
+    if _is_pg(conn):
+        op.create_table('external_knowledge_bindings',
+        sa.Column('id', models.types.StringUUID(), server_default=sa.text('uuid_generate_v4()'), nullable=False),
+        sa.Column('tenant_id', models.types.StringUUID(), nullable=False),
+        sa.Column('external_knowledge_api_id', models.types.StringUUID(), nullable=False),
+        sa.Column('dataset_id', models.types.StringUUID(), nullable=False),
+        sa.Column('external_knowledge_id', sa.Text(), nullable=False),
+        sa.Column('created_by', models.types.StringUUID(), nullable=False),
+        sa.Column('created_at', sa.DateTime(), server_default=sa.text('CURRENT_TIMESTAMP(0)'), nullable=False),
+        sa.Column('updated_by', models.types.StringUUID(), nullable=True),
+        sa.Column('updated_at', sa.DateTime(), server_default=sa.text('CURRENT_TIMESTAMP(0)'), nullable=False),
+        sa.PrimaryKeyConstraint('id', name='external_knowledge_bindings_pkey')
+        )
+    else:
+        op.create_table('external_knowledge_bindings',
+        sa.Column('id', models.types.StringUUID(), default=lambda: str(uuid4()), nullable=False),
+        sa.Column('tenant_id', models.types.StringUUID(), nullable=False),
+        sa.Column('external_knowledge_api_id', models.types.StringUUID(), nullable=False),
+        sa.Column('dataset_id', models.types.StringUUID(), nullable=False),
+        sa.Column('external_knowledge_id', sa.String(length=512), nullable=False),
+        sa.Column('created_by', models.types.StringUUID(), nullable=False),
+        sa.Column('created_at', sa.DateTime(), server_default=sa.func.current_timestamp(), nullable=False),
+        sa.Column('updated_by', models.types.StringUUID(), nullable=True),
+        sa.Column('updated_at', sa.DateTime(), server_default=sa.func.current_timestamp(), nullable=False),
+        sa.PrimaryKeyConstraint('id', name='external_knowledge_bindings_pkey')
+        )
+    
     with op.batch_alter_table('external_knowledge_bindings', schema=None) as batch_op:
         batch_op.create_index('external_knowledge_bindings_dataset_idx', ['dataset_id'], unique=False)
         batch_op.create_index('external_knowledge_bindings_external_knowledge_api_idx', ['external_knowledge_api_id'], unique=False)

api/migrations/versions/2024_10_10_0516-bbadea11becb_add_name_and_size_to_tool_files.py

@@ -16,6 +16,10 @@ branch_labels = None
 depends_on = None
 
 
+def _is_pg(conn):
+    return conn.dialect.name == "postgresql"
+
+
 def upgrade():
     def _has_name_or_size_column() -> bool:
         # We cannot access the database in offline mode, so assume
@@ -46,14 +50,26 @@ def upgrade():
     if _has_name_or_size_column():
         return
 
-    with op.batch_alter_table("tool_files", schema=None) as batch_op:
-        batch_op.add_column(sa.Column("name", sa.String(), nullable=True))
-        batch_op.add_column(sa.Column("size", sa.Integer(), nullable=True))
-    op.execute("UPDATE tool_files SET name = '' WHERE name IS NULL")
-    op.execute("UPDATE tool_files SET size = -1 WHERE size IS NULL")
-    with op.batch_alter_table("tool_files", schema=None) as batch_op:
-        batch_op.alter_column("name", existing_type=sa.String(), nullable=False)
-        batch_op.alter_column("size", existing_type=sa.Integer(), nullable=False)
+    if _is_pg(conn):
+        # PostgreSQL: Keep original syntax
+        with op.batch_alter_table("tool_files", schema=None) as batch_op:
+            batch_op.add_column(sa.Column("name", sa.String(), nullable=True))
+            batch_op.add_column(sa.Column("size", sa.Integer(), nullable=True))
+        op.execute("UPDATE tool_files SET name = '' WHERE name IS NULL")
+        op.execute("UPDATE tool_files SET size = -1 WHERE size IS NULL")
+        with op.batch_alter_table("tool_files", schema=None) as batch_op:
+            batch_op.alter_column("name", existing_type=sa.String(), nullable=False)
+            batch_op.alter_column("size", existing_type=sa.Integer(), nullable=False)
+    else:
+        # MySQL: Use compatible syntax
+        with op.batch_alter_table("tool_files", schema=None) as batch_op:
+            batch_op.add_column(sa.Column("name", sa.String(length=255), nullable=True))
+            batch_op.add_column(sa.Column("size", sa.Integer(), nullable=True))
+        op.execute("UPDATE tool_files SET name = '' WHERE name IS NULL")
+        op.execute("UPDATE tool_files SET size = -1 WHERE size IS NULL")
+        with op.batch_alter_table("tool_files", schema=None) as batch_op:
+            batch_op.alter_column("name", existing_type=sa.String(length=255), nullable=False)
+            batch_op.alter_column("size", existing_type=sa.Integer(), nullable=False)
     # ### end Alembic commands ###
 
 

api/migrations/versions/2024_10_22_0959-43fa78bc3b7d_add_white_list.py

@@ -9,6 +9,11 @@ from alembic import op
 import models as models
 import sqlalchemy as sa
 from sqlalchemy.dialects import postgresql
+from uuid import uuid4
+
+
+def _is_pg(conn):
+    return conn.dialect.name == "postgresql"
 
 # revision identifiers, used by Alembic.
 revision = '43fa78bc3b7d'
@@ -19,13 +24,25 @@ depends_on = None
 
 def upgrade():
     # ### commands auto generated by Alembic - please adjust! ###
-    op.create_table('whitelists',
-    sa.Column('id', models.types.StringUUID(), server_default=sa.text('uuid_generate_v4()'), nullable=False),
-    sa.Column('tenant_id', models.types.StringUUID(), nullable=True),
-    sa.Column('category', sa.String(length=255), nullable=False),
-    sa.Column('created_at', sa.DateTime(), server_default=sa.text('CURRENT_TIMESTAMP(0)'), nullable=False),
-    sa.PrimaryKeyConstraint('id', name='whitelists_pkey')
-    )
+    conn = op.get_bind()
+    
+    if _is_pg(conn):
+        op.create_table('whitelists',
+        sa.Column('id', models.types.StringUUID(), server_default=sa.text('uuid_generate_v4()'), nullable=False),
+        sa.Column('tenant_id', models.types.StringUUID(), nullable=True),
+        sa.Column('category', sa.String(length=255), nullable=False),
+        sa.Column('created_at', sa.DateTime(), server_default=sa.text('CURRENT_TIMESTAMP(0)'), nullable=False),
+        sa.PrimaryKeyConstraint('id', name='whitelists_pkey')
+        )
+    else:
+        op.create_table('whitelists',
+        sa.Column('id', models.types.StringUUID(), default=lambda: str(uuid4()), nullable=False),
+        sa.Column('tenant_id', models.types.StringUUID(), nullable=True),
+        sa.Column('category', sa.String(length=255), nullable=False),
+        sa.Column('created_at', sa.DateTime(), server_default=sa.func.current_timestamp(), nullable=False),
+        sa.PrimaryKeyConstraint('id', name='whitelists_pkey')
+        )
+    
     with op.batch_alter_table('whitelists', schema=None) as batch_op:
         batch_op.create_index('whitelists_tenant_idx', ['tenant_id'], unique=False)
 

api/migrations/versions/2024_10_28_0720-08ec4f75af5e_add_tenant_plugin_permisisons.py

@@ -9,6 +9,11 @@ from alembic import op
 import models as models
 import sqlalchemy as sa
 from sqlalchemy.dialects import postgresql
+from uuid import uuid4
+
+
+def _is_pg(conn):
+    return conn.dialect.name == "postgresql"
 
 # revision identifiers, used by Alembic.
 revision = '08ec4f75af5e'
@@ -19,14 +24,26 @@ depends_on = None
 
 def upgrade():
     # ### commands auto generated by Alembic - please adjust! ###
-    op.create_table('account_plugin_permissions',
-    sa.Column('id', models.types.StringUUID(), server_default=sa.text('uuid_generate_v4()'), nullable=False),
-    sa.Column('tenant_id', models.types.StringUUID(), nullable=False),
-    sa.Column('install_permission', sa.String(length=16), server_default='everyone', nullable=False),
-    sa.Column('debug_permission', sa.String(length=16), server_default='noone', nullable=False),
-    sa.PrimaryKeyConstraint('id', name='account_plugin_permission_pkey'),
-    sa.UniqueConstraint('tenant_id', name='unique_tenant_plugin')
-    )
+    conn = op.get_bind()
+    
+    if _is_pg(conn):
+        op.create_table('account_plugin_permissions',
+        sa.Column('id', models.types.StringUUID(), server_default=sa.text('uuid_generate_v4()'), nullable=False),
+        sa.Column('tenant_id', models.types.StringUUID(), nullable=False),
+        sa.Column('install_permission', sa.String(length=16), server_default='everyone', nullable=False),
+        sa.Column('debug_permission', sa.String(length=16), server_default='noone', nullable=False),
+        sa.PrimaryKeyConstraint('id', name='account_plugin_permission_pkey'),
+        sa.UniqueConstraint('tenant_id', name='unique_tenant_plugin')
+        )
+    else:
+        op.create_table('account_plugin_permissions',
+        sa.Column('id', models.types.StringUUID(), default=lambda: str(uuid4()), nullable=False),
+        sa.Column('tenant_id', models.types.StringUUID(), nullable=False),
+        sa.Column('install_permission', sa.String(length=16), server_default='everyone', nullable=False),
+        sa.Column('debug_permission', sa.String(length=16), server_default='noone', nullable=False),
+        sa.PrimaryKeyConstraint('id', name='account_plugin_permission_pkey'),
+        sa.UniqueConstraint('tenant_id', name='unique_tenant_plugin')
+        )
 
     # ### end Alembic commands ###
 

api/migrations/versions/2024_11_01_0540-f4d7ce70a7ca_update_upload_files_source_url.py

@@ -10,6 +10,10 @@ import models as models
 import sqlalchemy as sa
 from sqlalchemy.dialects import postgresql
 
+
+def _is_pg(conn):
+    return conn.dialect.name == "postgresql"
+
 # revision identifiers, used by Alembic.
 revision = 'f4d7ce70a7ca'
 down_revision = '93ad8c19c40b'
@@ -19,23 +23,43 @@ depends_on = None
 
 def upgrade():
     # ### commands auto generated by Alembic - please adjust! ###
-    with op.batch_alter_table('upload_files', schema=None) as batch_op:
-        batch_op.alter_column('source_url',
-               existing_type=sa.VARCHAR(length=255),
-               type_=sa.TEXT(),
-               existing_nullable=False,
-               existing_server_default=sa.text("''::character varying"))
+    conn = op.get_bind()
+    
+    if _is_pg(conn):
+        with op.batch_alter_table('upload_files', schema=None) as batch_op:
+            batch_op.alter_column('source_url',
+                   existing_type=sa.VARCHAR(length=255),
+                   type_=sa.TEXT(),
+                   existing_nullable=False,
+                   existing_server_default=sa.text("''::character varying"))
+    else:
+        with op.batch_alter_table('upload_files', schema=None) as batch_op:
+            batch_op.alter_column('source_url',
+                   existing_type=sa.VARCHAR(length=255),
+                   type_=models.types.LongText(),
+                   existing_nullable=False,
+                   existing_default=sa.text("''"))
 
     # ### end Alembic commands ###
 
 
 def downgrade():
     # ### commands auto generated by Alembic - please adjust! ###
-    with op.batch_alter_table('upload_files', schema=None) as batch_op:
-        batch_op.alter_column('source_url',
-               existing_type=sa.TEXT(),
-               type_=sa.VARCHAR(length=255),
-               existing_nullable=False,
-               existing_server_default=sa.text("''::character varying"))
+    conn = op.get_bind()
+    
+    if _is_pg(conn):
+        with op.batch_alter_table('upload_files', schema=None) as batch_op:
+            batch_op.alter_column('source_url',
+                   existing_type=sa.TEXT(),
+                   type_=sa.VARCHAR(length=255),
+                   existing_nullable=False,
+                   existing_server_default=sa.text("''::character varying"))
+    else:
+        with op.batch_alter_table('upload_files', schema=None) as batch_op:
+            batch_op.alter_column('source_url',
+                   existing_type=models.types.LongText(),
+                   type_=sa.VARCHAR(length=255),
+                   existing_nullable=False,
+                   existing_default=sa.text("''"))
 
     # ### end Alembic commands ###

api/migrations/versions/2024_11_01_0622-d07474999927_update_type_of_custom_disclaimer_to_text.py

@@ -7,6 +7,9 @@ Create Date: 2024-11-01 06:22:27.981398
 """
 from alembic import op
 import models as models
+
+def _is_pg(conn):
+    return conn.dialect.name == "postgresql"
 import sqlalchemy as sa
 from sqlalchemy.dialects import postgresql
 
@@ -19,49 +22,91 @@ depends_on = None
 
 def upgrade():
     # ### commands auto generated by Alembic - please adjust! ###
+    conn = op.get_bind()
+    
     op.execute("UPDATE recommended_apps SET custom_disclaimer = '' WHERE custom_disclaimer IS NULL")
     op.execute("UPDATE sites SET custom_disclaimer = '' WHERE custom_disclaimer IS NULL")
     op.execute("UPDATE tool_api_providers SET custom_disclaimer = '' WHERE custom_disclaimer IS NULL")
 
-    with op.batch_alter_table('recommended_apps', schema=None) as batch_op:
-        batch_op.alter_column('custom_disclaimer',
-               existing_type=sa.VARCHAR(length=255),
-               type_=sa.TEXT(),
-               nullable=False)
+    if _is_pg(conn):
+        with op.batch_alter_table('recommended_apps', schema=None) as batch_op:
+            batch_op.alter_column('custom_disclaimer',
+                   existing_type=sa.VARCHAR(length=255),
+                   type_=sa.TEXT(),
+                   nullable=False)
 
-    with op.batch_alter_table('sites', schema=None) as batch_op:
-        batch_op.alter_column('custom_disclaimer',
-               existing_type=sa.VARCHAR(length=255),
-               type_=sa.TEXT(),
-               nullable=False)
+        with op.batch_alter_table('sites', schema=None) as batch_op:
+            batch_op.alter_column('custom_disclaimer',
+                   existing_type=sa.VARCHAR(length=255),
+                   type_=sa.TEXT(),
+                   nullable=False)
 
-    with op.batch_alter_table('tool_api_providers', schema=None) as batch_op:
-        batch_op.alter_column('custom_disclaimer',
-               existing_type=sa.VARCHAR(length=255),
-               type_=sa.TEXT(),
-               nullable=False)
+        with op.batch_alter_table('tool_api_providers', schema=None) as batch_op:
+            batch_op.alter_column('custom_disclaimer',
+                   existing_type=sa.VARCHAR(length=255),
+                   type_=sa.TEXT(),
+                   nullable=False)
+    else:
+        with op.batch_alter_table('recommended_apps', schema=None) as batch_op:
+            batch_op.alter_column('custom_disclaimer',
+                   existing_type=sa.VARCHAR(length=255),
+                   type_=models.types.LongText(),
+                   nullable=False)
+
+        with op.batch_alter_table('sites', schema=None) as batch_op:
+            batch_op.alter_column('custom_disclaimer',
+                   existing_type=sa.VARCHAR(length=255),
+                   type_=models.types.LongText(),
+                   nullable=False)
+
+        with op.batch_alter_table('tool_api_providers', schema=None) as batch_op:
+            batch_op.alter_column('custom_disclaimer',
+                   existing_type=sa.VARCHAR(length=255),
+                   type_=models.types.LongText(),
+                   nullable=False)
 
     # ### end Alembic commands ###
 
 
 def downgrade():
     # ### commands auto generated by Alembic - please adjust! ###
-    with op.batch_alter_table('tool_api_providers', schema=None) as batch_op:
-        batch_op.alter_column('custom_disclaimer',
-               existing_type=sa.TEXT(),
-               type_=sa.VARCHAR(length=255),
-               nullable=True)
+    conn = op.get_bind()
+    
+    if _is_pg(conn):
+        with op.batch_alter_table('tool_api_providers', schema=None) as batch_op:
+            batch_op.alter_column('custom_disclaimer',
+                   existing_type=sa.TEXT(),
+                   type_=sa.VARCHAR(length=255),
+                   nullable=True)
 
-    with op.batch_alter_table('sites', schema=None) as batch_op:
-        batch_op.alter_column('custom_disclaimer',
-               existing_type=sa.TEXT(),
-               type_=sa.VARCHAR(length=255),
-               nullable=True)
+        with op.batch_alter_table('sites', schema=None) as batch_op:
+            batch_op.alter_column('custom_disclaimer',
+                   existing_type=sa.TEXT(),
+                   type_=sa.VARCHAR(length=255),
+                   nullable=True)
 
-    with op.batch_alter_table('recommended_apps', schema=None) as batch_op:
-        batch_op.alter_column('custom_disclaimer',
-               existing_type=sa.TEXT(),
-               type_=sa.VARCHAR(length=255),
-               nullable=True)
+        with op.batch_alter_table('recommended_apps', schema=None) as batch_op:
+            batch_op.alter_column('custom_disclaimer',
+                   existing_type=sa.TEXT(),
+                   type_=sa.VARCHAR(length=255),
+                   nullable=True)
+    else:
+        with op.batch_alter_table('tool_api_providers', schema=None) as batch_op:
+            batch_op.alter_column('custom_disclaimer',
+                   existing_type=models.types.LongText(),
+                   type_=sa.VARCHAR(length=255),
+                   nullable=True)
+
+        with op.batch_alter_table('sites', schema=None) as batch_op:
+            batch_op.alter_column('custom_disclaimer',
+                   existing_type=models.types.LongText(),
+                   type_=sa.VARCHAR(length=255),
+                   nullable=True)
+
+        with op.batch_alter_table('recommended_apps', schema=None) as batch_op:
+            batch_op.alter_column('custom_disclaimer',
+                   existing_type=models.types.LongText(),
+                   type_=sa.VARCHAR(length=255),
+                   nullable=True)
 
     # ### end Alembic commands ###

api/migrations/versions/2024_11_01_0623-09a8d1878d9b_update_workflows_graph_features_and_.py

@@ -10,6 +10,10 @@ import models as models
 import sqlalchemy as sa
 from sqlalchemy.dialects import postgresql
 
+
+def _is_pg(conn):
+    return conn.dialect.name == "postgresql"
+
 # revision identifiers, used by Alembic.
 revision = '09a8d1878d9b'
 down_revision = 'd07474999927'
@@ -19,55 +23,103 @@ depends_on = None
 
 def upgrade():
     # ### commands auto generated by Alembic - please adjust! ###
-    with op.batch_alter_table('conversations', schema=None) as batch_op:
-        batch_op.alter_column('inputs',
-               existing_type=postgresql.JSON(astext_type=sa.Text()),
-               nullable=False)
+    conn = op.get_bind()
+    
+    if _is_pg(conn):
+        with op.batch_alter_table('conversations', schema=None) as batch_op:
+            batch_op.alter_column('inputs',
+                   existing_type=postgresql.JSON(astext_type=sa.Text()),
+                   nullable=False)
 
-    with op.batch_alter_table('messages', schema=None) as batch_op:
-        batch_op.alter_column('inputs',
-               existing_type=postgresql.JSON(astext_type=sa.Text()),
-               nullable=False)
+        with op.batch_alter_table('messages', schema=None) as batch_op:
+            batch_op.alter_column('inputs',
+                   existing_type=postgresql.JSON(astext_type=sa.Text()),
+                   nullable=False)
+    else:
+        with op.batch_alter_table('conversations', schema=None) as batch_op:
+            batch_op.alter_column('inputs',
+                   existing_type=sa.JSON(),
+                   nullable=False)
+
+        with op.batch_alter_table('messages', schema=None) as batch_op:
+            batch_op.alter_column('inputs',
+                   existing_type=sa.JSON(),
+                   nullable=False)
 
     op.execute("UPDATE workflows SET updated_at = created_at WHERE updated_at IS NULL")
     op.execute("UPDATE workflows SET graph = '' WHERE graph IS NULL")
     op.execute("UPDATE workflows SET features = '' WHERE features IS NULL")
-
-    with op.batch_alter_table('workflows', schema=None) as batch_op:
-        batch_op.alter_column('graph',
-               existing_type=sa.TEXT(),
-               nullable=False)
-        batch_op.alter_column('features',
-               existing_type=sa.TEXT(),
-               nullable=False)
-        batch_op.alter_column('updated_at',
-               existing_type=postgresql.TIMESTAMP(),
-               nullable=False)
-
+    if _is_pg(conn):
+        with op.batch_alter_table('workflows', schema=None) as batch_op:
+            batch_op.alter_column('graph',
+                existing_type=sa.TEXT(),
+                nullable=False)
+            batch_op.alter_column('features',
+                existing_type=sa.TEXT(),
+                nullable=False)
+            batch_op.alter_column('updated_at',
+                existing_type=postgresql.TIMESTAMP(),
+                nullable=False)
+    else:
+        with op.batch_alter_table('workflows', schema=None) as batch_op:
+            batch_op.alter_column('graph',
+                existing_type=models.types.LongText(),
+                nullable=False)
+            batch_op.alter_column('features',
+                existing_type=models.types.LongText(),
+                nullable=False)
+            batch_op.alter_column('updated_at',
+                existing_type=sa.TIMESTAMP(),
+                nullable=False)
     # ### end Alembic commands ###
 
 
 def downgrade():
     # ### commands auto generated by Alembic - please adjust! ###
-    with op.batch_alter_table('workflows', schema=None) as batch_op:
-        batch_op.alter_column('updated_at',
-               existing_type=postgresql.TIMESTAMP(),
-               nullable=True)
-        batch_op.alter_column('features',
-               existing_type=sa.TEXT(),
-               nullable=True)
-        batch_op.alter_column('graph',
-               existing_type=sa.TEXT(),
-               nullable=True)
+    conn = op.get_bind()
+    
+    if _is_pg(conn):
+        with op.batch_alter_table('workflows', schema=None) as batch_op:
+            batch_op.alter_column('updated_at',
+                existing_type=postgresql.TIMESTAMP(),
+                nullable=True)
+            batch_op.alter_column('features',
+                existing_type=sa.TEXT(),
+                nullable=True)
+            batch_op.alter_column('graph',
+                existing_type=sa.TEXT(),
+                nullable=True)
+    else:
+        with op.batch_alter_table('workflows', schema=None) as batch_op:
+            batch_op.alter_column('updated_at',
+                existing_type=sa.TIMESTAMP(),
+                nullable=True)
+            batch_op.alter_column('features',
+                existing_type=models.types.LongText(),
+                nullable=True)
+            batch_op.alter_column('graph',
+                existing_type=models.types.LongText(),
+                nullable=True)
 
-    with op.batch_alter_table('messages', schema=None) as batch_op:
-        batch_op.alter_column('inputs',
-               existing_type=postgresql.JSON(astext_type=sa.Text()),
-               nullable=True)
+    if _is_pg(conn):
+        with op.batch_alter_table('messages', schema=None) as batch_op:
+            batch_op.alter_column('inputs',
+                   existing_type=postgresql.JSON(astext_type=sa.Text()),
+                   nullable=True)
 
-    with op.batch_alter_table('conversations', schema=None) as batch_op:
-        batch_op.alter_column('inputs',
-               existing_type=postgresql.JSON(astext_type=sa.Text()),
-               nullable=True)
+        with op.batch_alter_table('conversations', schema=None) as batch_op:
+            batch_op.alter_column('inputs',
+                   existing_type=postgresql.JSON(astext_type=sa.Text()),
+                   nullable=True)
+    else:
+        with op.batch_alter_table('messages', schema=None) as batch_op:
+            batch_op.alter_column('inputs',
+                   existing_type=sa.JSON(),
+                   nullable=True)
+
+        with op.batch_alter_table('conversations', schema=None) as batch_op:
+            batch_op.alter_column('inputs',
+                   existing_type=sa.JSON(),
+                   nullable=True)
 
     # ### end Alembic commands ###

api/migrations/versions/2024_11_22_0701-e19037032219_parent_child_index.py

@@ -8,6 +8,11 @@ Create Date: 2024-11-22 07:01:17.550037
 from alembic import op
 import models as models
 import sqlalchemy as sa
+from uuid import uuid4
+
+
+def _is_pg(conn):
+    return conn.dialect.name == "postgresql"
 
 
 # revision identifiers, used by Alembic.
@@ -19,27 +24,53 @@ depends_on = None
 
 def upgrade():
     # ### commands auto generated by Alembic - please adjust! ###
-    op.create_table('child_chunks',
-    sa.Column('id', models.types.StringUUID(), server_default=sa.text('uuid_generate_v4()'), nullable=False),
-    sa.Column('tenant_id', models.types.StringUUID(), nullable=False),
-    sa.Column('dataset_id', models.types.StringUUID(), nullable=False),
-    sa.Column('document_id', models.types.StringUUID(), nullable=False),
-    sa.Column('segment_id', models.types.StringUUID(), nullable=False),
-    sa.Column('position', sa.Integer(), nullable=False),
-    sa.Column('content', sa.Text(), nullable=False),
-    sa.Column('word_count', sa.Integer(), nullable=False),
-    sa.Column('index_node_id', sa.String(length=255), nullable=True),
-    sa.Column('index_node_hash', sa.String(length=255), nullable=True),
-    sa.Column('type', sa.String(length=255), server_default=sa.text("'automatic'::character varying"), nullable=False),
-    sa.Column('created_by', models.types.StringUUID(), nullable=False),
-    sa.Column('created_at', sa.DateTime(), server_default=sa.text('CURRENT_TIMESTAMP(0)'), nullable=False),
-    sa.Column('updated_by', models.types.StringUUID(), nullable=True),
-    sa.Column('updated_at', sa.DateTime(), server_default=sa.text('CURRENT_TIMESTAMP(0)'), nullable=False),
-    sa.Column('indexing_at', sa.DateTime(), nullable=True),
-    sa.Column('completed_at', sa.DateTime(), nullable=True),
-    sa.Column('error', sa.Text(), nullable=True),
-    sa.PrimaryKeyConstraint('id', name='child_chunk_pkey')
-    )
+    conn = op.get_bind()
+    
+    if _is_pg(conn):
+        op.create_table('child_chunks',
+        sa.Column('id', models.types.StringUUID(), server_default=sa.text('uuid_generate_v4()'), nullable=False),
+        sa.Column('tenant_id', models.types.StringUUID(), nullable=False),
+        sa.Column('dataset_id', models.types.StringUUID(), nullable=False),
+        sa.Column('document_id', models.types.StringUUID(), nullable=False),
+        sa.Column('segment_id', models.types.StringUUID(), nullable=False),
+        sa.Column('position', sa.Integer(), nullable=False),
+        sa.Column('content', sa.Text(), nullable=False),
+        sa.Column('word_count', sa.Integer(), nullable=False),
+        sa.Column('index_node_id', sa.String(length=255), nullable=True),
+        sa.Column('index_node_hash', sa.String(length=255), nullable=True),
+        sa.Column('type', sa.String(length=255), server_default=sa.text("'automatic'::character varying"), nullable=False),
+        sa.Column('created_by', models.types.StringUUID(), nullable=False),
+        sa.Column('created_at', sa.DateTime(), server_default=sa.text('CURRENT_TIMESTAMP(0)'), nullable=False),
+        sa.Column('updated_by', models.types.StringUUID(), nullable=True),
+        sa.Column('updated_at', sa.DateTime(), server_default=sa.text('CURRENT_TIMESTAMP(0)'), nullable=False),
+        sa.Column('indexing_at', sa.DateTime(), nullable=True),
+        sa.Column('completed_at', sa.DateTime(), nullable=True),
+        sa.Column('error', sa.Text(), nullable=True),
+        sa.PrimaryKeyConstraint('id', name='child_chunk_pkey')
+        )
+    else:
+        op.create_table('child_chunks',
+        sa.Column('id', models.types.StringUUID(), default=lambda: str(uuid4()), nullable=False),
+        sa.Column('tenant_id', models.types.StringUUID(), nullable=False),
+        sa.Column('dataset_id', models.types.StringUUID(), nullable=False),
+        sa.Column('document_id', models.types.StringUUID(), nullable=False),
+        sa.Column('segment_id', models.types.StringUUID(), nullable=False),
+        sa.Column('position', sa.Integer(), nullable=False),
+        sa.Column('content', models.types.LongText(), nullable=False),
+        sa.Column('word_count', sa.Integer(), nullable=False),
+        sa.Column('index_node_id', sa.String(length=255), nullable=True),
+        sa.Column('index_node_hash', sa.String(length=255), nullable=True),
+        sa.Column('type', sa.String(length=255), server_default=sa.text("'automatic'"), nullable=False),
+        sa.Column('created_by', models.types.StringUUID(), nullable=False),
+        sa.Column('created_at', sa.DateTime(), server_default=sa.func.current_timestamp(), nullable=False),
+        sa.Column('updated_by', models.types.StringUUID(), nullable=True),
+        sa.Column('updated_at', sa.DateTime(), server_default=sa.func.current_timestamp(), nullable=False),
+        sa.Column('indexing_at', sa.DateTime(), nullable=True),
+        sa.Column('completed_at', sa.DateTime(), nullable=True),
+        sa.Column('error', models.types.LongText(), nullable=True),
+        sa.PrimaryKeyConstraint('id', name='child_chunk_pkey')
+        )
+    
     with op.batch_alter_table('child_chunks', schema=None) as batch_op:
         batch_op.create_index('child_chunk_dataset_id_idx', ['tenant_id', 'dataset_id', 'document_id', 'segment_id', 'index_node_id'], unique=False)
 

api/migrations/versions/2024_12_19_1746-11b07f66c737_remove_unused_tool_providers.py

@@ -9,6 +9,11 @@ from alembic import op
 import models as models
 import sqlalchemy as sa
 from sqlalchemy.dialects import postgresql
+from uuid import uuid4
+
+
+def _is_pg(conn):
+    return conn.dialect.name == "postgresql"
 
 # revision identifiers, used by Alembic.
 revision = '11b07f66c737'
@@ -25,15 +30,30 @@ def upgrade():
 
 def downgrade():
     # ### commands auto generated by Alembic - please adjust! ###
-    op.create_table('tool_providers',
-    sa.Column('id', sa.UUID(), server_default=sa.text('uuid_generate_v4()'), autoincrement=False, nullable=False),
-    sa.Column('tenant_id', sa.UUID(), autoincrement=False, nullable=False),
-    sa.Column('tool_name', sa.VARCHAR(length=40), autoincrement=False, nullable=False),
-    sa.Column('encrypted_credentials', sa.TEXT(), autoincrement=False, nullable=True),
-    sa.Column('is_enabled', sa.BOOLEAN(), server_default=sa.text('false'), autoincrement=False, nullable=False),
-    sa.Column('created_at', postgresql.TIMESTAMP(), server_default=sa.text('CURRENT_TIMESTAMP(0)'), autoincrement=False, nullable=False),
-    sa.Column('updated_at', postgresql.TIMESTAMP(), server_default=sa.text('CURRENT_TIMESTAMP(0)'), autoincrement=False, nullable=False),
-    sa.PrimaryKeyConstraint('id', name='tool_provider_pkey'),
-    sa.UniqueConstraint('tenant_id', 'tool_name', name='unique_tool_provider_tool_name')
-    )
+    conn = op.get_bind()
+    
+    if _is_pg(conn):
+        op.create_table('tool_providers',
+        sa.Column('id', sa.UUID(), server_default=sa.text('uuid_generate_v4()'), autoincrement=False, nullable=False),
+        sa.Column('tenant_id', sa.UUID(), autoincrement=False, nullable=False),
+        sa.Column('tool_name', sa.VARCHAR(length=40), autoincrement=False, nullable=False),
+        sa.Column('encrypted_credentials', sa.TEXT(), autoincrement=False, nullable=True),
+        sa.Column('is_enabled', sa.BOOLEAN(), server_default=sa.text('false'), autoincrement=False, nullable=False),
+        sa.Column('created_at', postgresql.TIMESTAMP(), server_default=sa.text('CURRENT_TIMESTAMP(0)'), autoincrement=False, nullable=False),
+        sa.Column('updated_at', postgresql.TIMESTAMP(), server_default=sa.text('CURRENT_TIMESTAMP(0)'), autoincrement=False, nullable=False),
+        sa.PrimaryKeyConstraint('id', name='tool_provider_pkey'),
+        sa.UniqueConstraint('tenant_id', 'tool_name', name='unique_tool_provider_tool_name')
+        )
+    else:
+        op.create_table('tool_providers',
+        sa.Column('id', models.types.StringUUID(), default=lambda: str(uuid4()), autoincrement=False, nullable=False),
+        sa.Column('tenant_id', models.types.StringUUID(), autoincrement=False, nullable=False),
+        sa.Column('tool_name', sa.VARCHAR(length=40), autoincrement=False, nullable=False),
+        sa.Column('encrypted_credentials', models.types.LongText(), autoincrement=False, nullable=True),
+        sa.Column('is_enabled', sa.BOOLEAN(), server_default=sa.text('false'), autoincrement=False, nullable=False),
+        sa.Column('created_at', sa.TIMESTAMP(), server_default=sa.func.current_timestamp(), autoincrement=False, nullable=False),
+        sa.Column('updated_at', sa.TIMESTAMP(), server_default=sa.func.current_timestamp(), autoincrement=False, nullable=False),
+        sa.PrimaryKeyConstraint('id', name='tool_provider_pkey'),
+        sa.UniqueConstraint('tenant_id', 'tool_name', name='unique_tool_provider_tool_name')
+        )
     # ### end Alembic commands ###

api/migrations/versions/2024_12_25_1137-923752d42eb6_add_auto_disabled_dataset_logs.py

@@ -9,6 +9,11 @@ from alembic import op
 import models as models
 import sqlalchemy as sa
 from sqlalchemy.dialects import postgresql
+from uuid import uuid4
+
+
+def _is_pg(conn):
+    return conn.dialect.name == "postgresql"
 
 # revision identifiers, used by Alembic.
 revision = '923752d42eb6'
@@ -19,15 +24,29 @@ depends_on = None
 
 def upgrade():
     # ### commands auto generated by Alembic - please adjust! ###
-    op.create_table('dataset_auto_disable_logs',
-    sa.Column('id', models.types.StringUUID(), server_default=sa.text('uuid_generate_v4()'), nullable=False),
-    sa.Column('tenant_id', models.types.StringUUID(), nullable=False),
-    sa.Column('dataset_id', models.types.StringUUID(), nullable=False),
-    sa.Column('document_id', models.types.StringUUID(), nullable=False),
-    sa.Column('notified', sa.Boolean(), server_default=sa.text('false'), nullable=False),
-    sa.Column('created_at', sa.DateTime(), server_default=sa.text('CURRENT_TIMESTAMP(0)'), nullable=False),
-    sa.PrimaryKeyConstraint('id', name='dataset_auto_disable_log_pkey')
-    )
+    conn = op.get_bind()
+    
+    if _is_pg(conn):
+        op.create_table('dataset_auto_disable_logs',
+        sa.Column('id', models.types.StringUUID(), server_default=sa.text('uuid_generate_v4()'), nullable=False),
+        sa.Column('tenant_id', models.types.StringUUID(), nullable=False),
+        sa.Column('dataset_id', models.types.StringUUID(), nullable=False),
+        sa.Column('document_id', models.types.StringUUID(), nullable=False),
+        sa.Column('notified', sa.Boolean(), server_default=sa.text('false'), nullable=False),
+        sa.Column('created_at', sa.DateTime(), server_default=sa.text('CURRENT_TIMESTAMP(0)'), nullable=False),
+        sa.PrimaryKeyConstraint('id', name='dataset_auto_disable_log_pkey')
+        )
+    else:
+        op.create_table('dataset_auto_disable_logs',
+        sa.Column('id', models.types.StringUUID(), default=lambda: str(uuid4()), nullable=False),
+        sa.Column('tenant_id', models.types.StringUUID(), nullable=False),
+        sa.Column('dataset_id', models.types.StringUUID(), nullable=False),
+        sa.Column('document_id', models.types.StringUUID(), nullable=False),
+        sa.Column('notified', sa.Boolean(), server_default=sa.text('false'), nullable=False),
+        sa.Column('created_at', sa.DateTime(), server_default=sa.func.current_timestamp(), nullable=False),
+        sa.PrimaryKeyConstraint('id', name='dataset_auto_disable_log_pkey')
+        )
+    
     with op.batch_alter_table('dataset_auto_disable_logs', schema=None) as batch_op:
         batch_op.create_index('dataset_auto_disable_log_created_atx', ['created_at'], unique=False)
         batch_op.create_index('dataset_auto_disable_log_dataset_idx', ['dataset_id'], unique=False)

api/migrations/versions/2025_01_14_0617-f051706725cc_add_rate_limit_logs.py

@@ -9,6 +9,11 @@ from alembic import op
 import models as models
 import sqlalchemy as sa
 from sqlalchemy.dialects import postgresql
+from uuid import uuid4
+
+
+def _is_pg(conn):
+    return conn.dialect.name == "postgresql"
 
 # revision identifiers, used by Alembic.
 revision = 'f051706725cc'
@@ -19,14 +24,27 @@ depends_on = None
 
 def upgrade():
     # ### commands auto generated by Alembic - please adjust! ###
-    op.create_table('rate_limit_logs',
-    sa.Column('id', models.types.StringUUID(), server_default=sa.text('uuid_generate_v4()'), nullable=False),
-    sa.Column('tenant_id', models.types.StringUUID(), nullable=False),
-    sa.Column('subscription_plan', sa.String(length=255), nullable=False),
-    sa.Column('operation', sa.String(length=255), nullable=False),
-    sa.Column('created_at', sa.DateTime(), server_default=sa.text('CURRENT_TIMESTAMP(0)'), nullable=False),
-    sa.PrimaryKeyConstraint('id', name='rate_limit_log_pkey')
-    )
+    conn = op.get_bind()
+    
+    if _is_pg(conn):
+        op.create_table('rate_limit_logs',
+        sa.Column('id', models.types.StringUUID(), server_default=sa.text('uuid_generate_v4()'), nullable=False),
+        sa.Column('tenant_id', models.types.StringUUID(), nullable=False),
+        sa.Column('subscription_plan', sa.String(length=255), nullable=False),
+        sa.Column('operation', sa.String(length=255), nullable=False),
+        sa.Column('created_at', sa.DateTime(), server_default=sa.text('CURRENT_TIMESTAMP(0)'), nullable=False),
+        sa.PrimaryKeyConstraint('id', name='rate_limit_log_pkey')
+        )
+    else:
+        op.create_table('rate_limit_logs',
+        sa.Column('id', models.types.StringUUID(), default=lambda: str(uuid4()), nullable=False),
+        sa.Column('tenant_id', models.types.StringUUID(), nullable=False),
+        sa.Column('subscription_plan', sa.String(length=255), nullable=False),
+        sa.Column('operation', sa.String(length=255), nullable=False),
+        sa.Column('created_at', sa.DateTime(), server_default=sa.func.current_timestamp(), nullable=False),
+        sa.PrimaryKeyConstraint('id', name='rate_limit_log_pkey')
+        )
+    
     with op.batch_alter_table('rate_limit_logs', schema=None) as batch_op:
         batch_op.create_index('rate_limit_log_operation_idx', ['operation'], unique=False)
         batch_op.create_index('rate_limit_log_tenant_idx', ['tenant_id'], unique=False)

api/migrations/versions/2025_02_27_0917-d20049ed0af6_add_metadata_function.py

@@ -9,6 +9,11 @@ from alembic import op
 import models as models
 import sqlalchemy as sa
 from sqlalchemy.dialects import postgresql
+from uuid import uuid4
+
+
+def _is_pg(conn):
+    return conn.dialect.name == "postgresql"
 
 # revision identifiers, used by Alembic.
 revision = 'd20049ed0af6'
@@ -19,34 +24,68 @@ depends_on = None
 
 def upgrade():
     # ### commands auto generated by Alembic - please adjust! ###
-    op.create_table('dataset_metadata_bindings',
-    sa.Column('id', models.types.StringUUID(), server_default=sa.text('uuid_generate_v4()'), nullable=False),
-    sa.Column('tenant_id', models.types.StringUUID(), nullable=False),
-    sa.Column('dataset_id', models.types.StringUUID(), nullable=False),
-    sa.Column('metadata_id', models.types.StringUUID(), nullable=False),
-    sa.Column('document_id', models.types.StringUUID(), nullable=False),
-    sa.Column('created_at', sa.DateTime(), server_default=sa.text('CURRENT_TIMESTAMP'), nullable=False),
-    sa.Column('created_by', models.types.StringUUID(), nullable=False),
-    sa.PrimaryKeyConstraint('id', name='dataset_metadata_binding_pkey')
-    )
+    conn = op.get_bind()
+    
+    if _is_pg(conn):
+        # PostgreSQL: Keep original syntax
+        op.create_table('dataset_metadata_bindings',
+        sa.Column('id', models.types.StringUUID(), server_default=sa.text('uuid_generate_v4()'), nullable=False),
+        sa.Column('tenant_id', models.types.StringUUID(), nullable=False),
+        sa.Column('dataset_id', models.types.StringUUID(), nullable=False),
+        sa.Column('metadata_id', models.types.StringUUID(), nullable=False),
+        sa.Column('document_id', models.types.StringUUID(), nullable=False),
+        sa.Column('created_at', sa.DateTime(), server_default=sa.text('CURRENT_TIMESTAMP'), nullable=False),
+        sa.Column('created_by', models.types.StringUUID(), nullable=False),
+        sa.PrimaryKeyConstraint('id', name='dataset_metadata_binding_pkey')
+        )
+    else:
+        # MySQL: Use compatible syntax
+        op.create_table('dataset_metadata_bindings',
+        sa.Column('id', models.types.StringUUID(), default=lambda: str(uuid4()), nullable=False),
+        sa.Column('tenant_id', models.types.StringUUID(), nullable=False),
+        sa.Column('dataset_id', models.types.StringUUID(), nullable=False),
+        sa.Column('metadata_id', models.types.StringUUID(), nullable=False),
+        sa.Column('document_id', models.types.StringUUID(), nullable=False),
+        sa.Column('created_at', sa.DateTime(), server_default=sa.func.current_timestamp(), nullable=False),
+        sa.Column('created_by', models.types.StringUUID(), nullable=False),
+        sa.PrimaryKeyConstraint('id', name='dataset_metadata_binding_pkey')
+        )
+    
     with op.batch_alter_table('dataset_metadata_bindings', schema=None) as batch_op:
         batch_op.create_index('dataset_metadata_binding_dataset_idx', ['dataset_id'], unique=False)
         batch_op.create_index('dataset_metadata_binding_document_idx', ['document_id'], unique=False)
         batch_op.create_index('dataset_metadata_binding_metadata_idx', ['metadata_id'], unique=False)
         batch_op.create_index('dataset_metadata_binding_tenant_idx', ['tenant_id'], unique=False)
 
-    op.create_table('dataset_metadatas',
-    sa.Column('id', models.types.StringUUID(), server_default=sa.text('uuid_generate_v4()'), nullable=False),
-    sa.Column('tenant_id', models.types.StringUUID(), nullable=False),
-    sa.Column('dataset_id', models.types.StringUUID(), nullable=False),
-    sa.Column('type', sa.String(length=255), nullable=False),
-    sa.Column('name', sa.String(length=255), nullable=False),
-    sa.Column('created_at', sa.DateTime(), server_default=sa.text('CURRENT_TIMESTAMP(0)'), nullable=False),
-    sa.Column('updated_at', sa.DateTime(), server_default=sa.text('CURRENT_TIMESTAMP(0)'), nullable=False),
-    sa.Column('created_by', models.types.StringUUID(), nullable=False),
-    sa.Column('updated_by', models.types.StringUUID(), nullable=True),
-    sa.PrimaryKeyConstraint('id', name='dataset_metadata_pkey')
-    )
+    if _is_pg(conn):
+        # PostgreSQL: Keep original syntax
+        op.create_table('dataset_metadatas',
+        sa.Column('id', models.types.StringUUID(), server_default=sa.text('uuid_generate_v4()'), nullable=False),
+        sa.Column('tenant_id', models.types.StringUUID(), nullable=False),
+        sa.Column('dataset_id', models.types.StringUUID(), nullable=False),
+        sa.Column('type', sa.String(length=255), nullable=False),
+        sa.Column('name', sa.String(length=255), nullable=False),
+        sa.Column('created_at', sa.DateTime(), server_default=sa.text('CURRENT_TIMESTAMP(0)'), nullable=False),
+        sa.Column('updated_at', sa.DateTime(), server_default=sa.text('CURRENT_TIMESTAMP(0)'), nullable=False),
+        sa.Column('created_by', models.types.StringUUID(), nullable=False),
+        sa.Column('updated_by', models.types.StringUUID(), nullable=True),
+        sa.PrimaryKeyConstraint('id', name='dataset_metadata_pkey')
+        )
+    else:
+        # MySQL: Use compatible syntax
+        op.create_table('dataset_metadatas',
+        sa.Column('id', models.types.StringUUID(), default=lambda: str(uuid4()), nullable=False),
+        sa.Column('tenant_id', models.types.StringUUID(), nullable=False),
+        sa.Column('dataset_id', models.types.StringUUID(), nullable=False),
+        sa.Column('type', sa.String(length=255), nullable=False),
+        sa.Column('name', sa.String(length=255), nullable=False),
+        sa.Column('created_at', sa.DateTime(), server_default=sa.func.current_timestamp(), nullable=False),
+        sa.Column('updated_at', sa.DateTime(), server_default=sa.func.current_timestamp(), nullable=False),
+        sa.Column('created_by', models.types.StringUUID(), nullable=False),
+        sa.Column('updated_by', models.types.StringUUID(), nullable=True),
+        sa.PrimaryKeyConstraint('id', name='dataset_metadata_pkey')
+        )
+    
     with op.batch_alter_table('dataset_metadatas', schema=None) as batch_op:
         batch_op.create_index('dataset_metadata_dataset_idx', ['dataset_id'], unique=False)
         batch_op.create_index('dataset_metadata_tenant_idx', ['tenant_id'], unique=False)
@@ -54,23 +93,31 @@ def upgrade():
     with op.batch_alter_table('datasets', schema=None) as batch_op:
         batch_op.add_column(sa.Column('built_in_field_enabled', sa.Boolean(), server_default=sa.text('false'), nullable=False))
 
-    with op.batch_alter_table('documents', schema=None) as batch_op:
-        batch_op.alter_column('doc_metadata',
-               existing_type=postgresql.JSON(astext_type=sa.Text()),
-               type_=postgresql.JSONB(astext_type=sa.Text()),
-               existing_nullable=True)
-        batch_op.create_index('document_metadata_idx', ['doc_metadata'], unique=False, postgresql_using='gin')
+    if _is_pg(conn):
+        with op.batch_alter_table('documents', schema=None) as batch_op:
+            batch_op.alter_column('doc_metadata',
+                   existing_type=postgresql.JSON(astext_type=sa.Text()),
+                   type_=postgresql.JSONB(astext_type=sa.Text()),
+                   existing_nullable=True)
+            batch_op.create_index('document_metadata_idx', ['doc_metadata'], unique=False, postgresql_using='gin')
+    else:
+        pass
     # ### end Alembic commands ###
 
 
 def downgrade():
     # ### commands auto generated by Alembic - please adjust! ###
-    with op.batch_alter_table('documents', schema=None) as batch_op:
-        batch_op.drop_index('document_metadata_idx', postgresql_using='gin')
-        batch_op.alter_column('doc_metadata',
-               existing_type=postgresql.JSONB(astext_type=sa.Text()),
-               type_=postgresql.JSON(astext_type=sa.Text()),
-               existing_nullable=True)
+    conn = op.get_bind()
+    
+    if _is_pg(conn):
+        with op.batch_alter_table('documents', schema=None) as batch_op:
+            batch_op.drop_index('document_metadata_idx', postgresql_using='gin')
+            batch_op.alter_column('doc_metadata',
+                   existing_type=postgresql.JSONB(astext_type=sa.Text()),
+                   type_=postgresql.JSON(astext_type=sa.Text()),
+                   existing_nullable=True)
+    else:
+        pass
 
     with op.batch_alter_table('datasets', schema=None) as batch_op:
         batch_op.drop_column('built_in_field_enabled')

api/migrations/versions/2025_03_03_1436-ee79d9b1c156_add_marked_name_and_marked_comment_in_.py

@@ -17,10 +17,23 @@ branch_labels = None
 depends_on = None
 
 
+def _is_pg(conn):
+    return conn.dialect.name == "postgresql"
+
+
 def upgrade():
-    with op.batch_alter_table('workflows', schema=None) as batch_op:
-        batch_op.add_column(sa.Column('marked_name', sa.String(), nullable=False, server_default=''))
-        batch_op.add_column(sa.Column('marked_comment', sa.String(), nullable=False, server_default=''))
+    conn = op.get_bind()
+    
+    if _is_pg(conn):
+        # PostgreSQL: Keep original syntax
+        with op.batch_alter_table('workflows', schema=None) as batch_op:
+            batch_op.add_column(sa.Column('marked_name', sa.String(), nullable=False, server_default=''))
+            batch_op.add_column(sa.Column('marked_comment', sa.String(), nullable=False, server_default=''))
+    else:
+        # MySQL: Use compatible syntax
+        with op.batch_alter_table('workflows', schema=None) as batch_op:
+            batch_op.add_column(sa.Column('marked_name', sa.String(length=255), nullable=False, server_default=''))
+            batch_op.add_column(sa.Column('marked_comment', sa.String(length=255), nullable=False, server_default=''))
 
 
 def downgrade():

api/migrations/versions/2025_05_15_1531-2adcbe1f5dfb_add_workflowdraftvariable_model.py

@@ -8,9 +8,14 @@ Create Date: 2025-05-15 15:31:03.128680
 
 import sqlalchemy as sa
 from alembic import op
+from uuid import uuid4
 
 import models as models
 
+
+def _is_pg(conn):
+    return conn.dialect.name == "postgresql"
+
 # revision identifiers, used by Alembic.
 revision = "2adcbe1f5dfb"
 down_revision = "d28f2004b072"
@@ -20,24 +25,46 @@ depends_on = None
 
 def upgrade():
     # ### commands auto generated by Alembic - please adjust! ###
-    op.create_table(
-        "workflow_draft_variables",
-        sa.Column("id", models.types.StringUUID(), server_default=sa.text("uuid_generate_v4()"), nullable=False),
-        sa.Column("created_at", sa.DateTime(), server_default=sa.text("CURRENT_TIMESTAMP"), nullable=False),
-        sa.Column("updated_at", sa.DateTime(), server_default=sa.text("CURRENT_TIMESTAMP"), nullable=False),
-        sa.Column("app_id", models.types.StringUUID(), nullable=False),
-        sa.Column("last_edited_at", sa.DateTime(), nullable=True),
-        sa.Column("node_id", sa.String(length=255), nullable=False),
-        sa.Column("name", sa.String(length=255), nullable=False),
-        sa.Column("description", sa.String(length=255), nullable=False),
-        sa.Column("selector", sa.String(length=255), nullable=False),
-        sa.Column("value_type", sa.String(length=20), nullable=False),
-        sa.Column("value", sa.Text(), nullable=False),
-        sa.Column("visible", sa.Boolean(), nullable=False),
-        sa.Column("editable", sa.Boolean(), nullable=False),
-        sa.PrimaryKeyConstraint("id", name=op.f("workflow_draft_variables_pkey")),
-        sa.UniqueConstraint("app_id", "node_id", "name", name=op.f("workflow_draft_variables_app_id_key")),
-    )
+    conn = op.get_bind()
+    
+    if _is_pg(conn):
+        op.create_table(
+            "workflow_draft_variables",
+            sa.Column("id", models.types.StringUUID(), server_default=sa.text("uuid_generate_v4()"), nullable=False),
+            sa.Column("created_at", sa.DateTime(), server_default=sa.text("CURRENT_TIMESTAMP"), nullable=False),
+            sa.Column("updated_at", sa.DateTime(), server_default=sa.text("CURRENT_TIMESTAMP"), nullable=False),
+            sa.Column("app_id", models.types.StringUUID(), nullable=False),
+            sa.Column("last_edited_at", sa.DateTime(), nullable=True),
+            sa.Column("node_id", sa.String(length=255), nullable=False),
+            sa.Column("name", sa.String(length=255), nullable=False),
+            sa.Column("description", sa.String(length=255), nullable=False),
+            sa.Column("selector", sa.String(length=255), nullable=False),
+            sa.Column("value_type", sa.String(length=20), nullable=False),
+            sa.Column("value", sa.Text(), nullable=False),
+            sa.Column("visible", sa.Boolean(), nullable=False),
+            sa.Column("editable", sa.Boolean(), nullable=False),
+            sa.PrimaryKeyConstraint("id", name=op.f("workflow_draft_variables_pkey")),
+            sa.UniqueConstraint("app_id", "node_id", "name", name=op.f("workflow_draft_variables_app_id_key")),
+        )
+    else:
+        op.create_table(
+            "workflow_draft_variables",
+            sa.Column("id", models.types.StringUUID(), default=lambda: str(uuid4()), nullable=False),
+            sa.Column("created_at", sa.DateTime(), server_default=sa.func.current_timestamp(), nullable=False),
+            sa.Column("updated_at", sa.DateTime(), server_default=sa.func.current_timestamp(), nullable=False),
+            sa.Column("app_id", models.types.StringUUID(), nullable=False),
+            sa.Column("last_edited_at", sa.DateTime(), nullable=True),
+            sa.Column("node_id", sa.String(length=255), nullable=False),
+            sa.Column("name", sa.String(length=255), nullable=False),
+            sa.Column("description", sa.String(length=255), nullable=False),
+            sa.Column("selector", sa.String(length=255), nullable=False),
+            sa.Column("value_type", sa.String(length=20), nullable=False),
+            sa.Column("value", models.types.LongText(), nullable=False),
+            sa.Column("visible", sa.Boolean(), nullable=False),
+            sa.Column("editable", sa.Boolean(), nullable=False),
+            sa.PrimaryKeyConstraint("id", name=op.f("workflow_draft_variables_pkey")),
+            sa.UniqueConstraint("app_id", "node_id", "name", name=op.f("workflow_draft_variables_app_id_key")),
+        )
 
     # ### end Alembic commands ###
 

api/migrations/versions/2025_06_06_1424-4474872b0ee6_workflow_draft_varaibles_add_node_execution_id.py

@@ -7,6 +7,10 @@ Create Date: 2025-06-06 14:24:44.213018
 """
 from alembic import op
 import models as models
+
+
+def _is_pg(conn):
+    return conn.dialect.name == "postgresql"
 import sqlalchemy as sa
 
 
@@ -18,19 +22,30 @@ depends_on = None
 
 
 def upgrade():
-    # `CREATE INDEX CONCURRENTLY` cannot run within a transaction, so use the `autocommit_block`
-    # context manager to wrap the index creation statement.
-    # Reference:
-    #
-    # - https://www.postgresql.org/docs/current/sql-createindex.html#:~:text=Another%20difference%20is,CREATE%20INDEX%20CONCURRENTLY%20cannot.
-    # - https://alembic.sqlalchemy.org/en/latest/api/runtime.html#alembic.runtime.migration.MigrationContext.autocommit_block
-    with op.get_context().autocommit_block():
+    # ### commands auto generated by Alembic - please adjust! ###
+    conn = op.get_bind()
+    
+    if _is_pg(conn):
+        # `CREATE INDEX CONCURRENTLY` cannot run within a transaction, so use the `autocommit_block`
+        # context manager to wrap the index creation statement.
+        # Reference:
+        #
+        # - https://www.postgresql.org/docs/current/sql-createindex.html#:~:text=Another%20difference%20is,CREATE%20INDEX%20CONCURRENTLY%20cannot.
+        # - https://alembic.sqlalchemy.org/en/latest/api/runtime.html#alembic.runtime.migration.MigrationContext.autocommit_block
+        with op.get_context().autocommit_block():
+            op.create_index(
+                op.f('workflow_node_executions_tenant_id_idx'),
+                "workflow_node_executions",
+                ['tenant_id', 'workflow_id', 'node_id', sa.literal_column('created_at DESC')],
+                unique=False,
+                postgresql_concurrently=True,
+            )
+    else:
         op.create_index(
             op.f('workflow_node_executions_tenant_id_idx'),
             "workflow_node_executions",
             ['tenant_id', 'workflow_id', 'node_id', sa.literal_column('created_at DESC')],
             unique=False,
-            postgresql_concurrently=True,
         )
 
     with op.batch_alter_table('workflow_draft_variables', schema=None) as batch_op:
@@ -51,8 +66,13 @@ def downgrade():
     # Reference:
     #
     # https://www.postgresql.org/docs/current/sql-createindex.html#:~:text=Another%20difference%20is,CREATE%20INDEX%20CONCURRENTLY%20cannot.
-    with op.get_context().autocommit_block():
-        op.drop_index(op.f('workflow_node_executions_tenant_id_idx'), postgresql_concurrently=True)
+    conn = op.get_bind()
+    
+    if _is_pg(conn):
+        with op.get_context().autocommit_block():
+            op.drop_index(op.f('workflow_node_executions_tenant_id_idx'), postgresql_concurrently=True)
+    else:
+        op.drop_index(op.f('workflow_node_executions_tenant_id_idx'))
 
     with op.batch_alter_table('workflow_draft_variables', schema=None) as batch_op:
         batch_op.drop_column('node_execution_id')

api/migrations/versions/2025_06_25_0936-58eb7bdb93fe_add_mcp_server_tool_and_app_server.py

@@ -8,6 +8,11 @@ Create Date: 2025-06-25 09:36:07.510570
 from alembic import op
 import models as models
 import sqlalchemy as sa
+from uuid import uuid4
+
+
+def _is_pg(conn):
+    return conn.dialect.name == "postgresql"
 
 
 # revision identifiers, used by Alembic.
@@ -19,40 +24,80 @@ depends_on = None
 
 def upgrade():
     # ### commands auto generated by Alembic - please adjust! ###
-    op.create_table('app_mcp_servers',
-    sa.Column('id', models.types.StringUUID(), server_default=sa.text('uuid_generate_v4()'), nullable=False),
-    sa.Column('tenant_id', models.types.StringUUID(), nullable=False),
-    sa.Column('app_id', models.types.StringUUID(), nullable=False),
-    sa.Column('name', sa.String(length=255), nullable=False),
-    sa.Column('description', sa.String(length=255), nullable=False),
-    sa.Column('server_code', sa.String(length=255), nullable=False),
-    sa.Column('status', sa.String(length=255), server_default=sa.text("'normal'::character varying"), nullable=False),
-    sa.Column('parameters', sa.Text(), nullable=False),
-    sa.Column('created_at', sa.DateTime(), server_default=sa.text('CURRENT_TIMESTAMP'), nullable=False),
-    sa.Column('updated_at', sa.DateTime(), server_default=sa.text('CURRENT_TIMESTAMP'), nullable=False),
-    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')
-    )
-    op.create_table('tool_mcp_providers',
-    sa.Column('id', models.types.StringUUID(), server_default=sa.text('uuid_generate_v4()'), nullable=False),
-    sa.Column('name', sa.String(length=40), nullable=False),
-    sa.Column('server_identifier', sa.String(length=24), nullable=False),
-    sa.Column('server_url', sa.Text(), nullable=False),
-    sa.Column('server_url_hash', sa.String(length=64), nullable=False),
-    sa.Column('icon', sa.String(length=255), nullable=True),
-    sa.Column('tenant_id', models.types.StringUUID(), nullable=False),
-    sa.Column('user_id', models.types.StringUUID(), nullable=False),
-    sa.Column('encrypted_credentials', sa.Text(), nullable=True),
-    sa.Column('authed', sa.Boolean(), nullable=False),
-    sa.Column('tools', sa.Text(), nullable=False),
-    sa.Column('created_at', sa.DateTime(), server_default=sa.text('CURRENT_TIMESTAMP(0)'), nullable=False),
-    sa.Column('updated_at', sa.DateTime(), server_default=sa.text('CURRENT_TIMESTAMP(0)'), nullable=False),
-    sa.PrimaryKeyConstraint('id', name='tool_mcp_provider_pkey'),
-    sa.UniqueConstraint('tenant_id', 'name', name='unique_mcp_provider_name'),
-    sa.UniqueConstraint('tenant_id', 'server_identifier', name='unique_mcp_provider_server_identifier'),
-    sa.UniqueConstraint('tenant_id', 'server_url_hash', name='unique_mcp_provider_server_url')
-    )
+    conn = op.get_bind()
+    
+    if _is_pg(conn):
+        op.create_table('app_mcp_servers',
+        sa.Column('id', models.types.StringUUID(), server_default=sa.text('uuid_generate_v4()'), nullable=False),
+        sa.Column('tenant_id', models.types.StringUUID(), nullable=False),
+        sa.Column('app_id', models.types.StringUUID(), nullable=False),
+        sa.Column('name', sa.String(length=255), nullable=False),
+        sa.Column('description', sa.String(length=255), nullable=False),
+        sa.Column('server_code', sa.String(length=255), nullable=False),
+        sa.Column('status', sa.String(length=255), server_default=sa.text("'normal'::character varying"), nullable=False),
+        sa.Column('parameters', sa.Text(), nullable=False),
+        sa.Column('created_at', sa.DateTime(), server_default=sa.text('CURRENT_TIMESTAMP'), nullable=False),
+        sa.Column('updated_at', sa.DateTime(), server_default=sa.text('CURRENT_TIMESTAMP'), nullable=False),
+        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')
+        )
+    else:
+        op.create_table('app_mcp_servers',
+        sa.Column('id', models.types.StringUUID(), default=lambda: str(uuid4()), nullable=False),
+        sa.Column('tenant_id', models.types.StringUUID(), nullable=False),
+        sa.Column('app_id', models.types.StringUUID(), nullable=False),
+        sa.Column('name', sa.String(length=255), nullable=False),
+        sa.Column('description', sa.String(length=255), nullable=False),
+        sa.Column('server_code', sa.String(length=255), nullable=False),
+        sa.Column('status', sa.String(length=255), server_default=sa.text("'normal'"), nullable=False),
+        sa.Column('parameters', models.types.LongText(), nullable=False),
+        sa.Column('created_at', sa.DateTime(), server_default=sa.func.current_timestamp(), nullable=False),
+        sa.Column('updated_at', sa.DateTime(), server_default=sa.func.current_timestamp(), nullable=False),
+        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')
+        )
+    if _is_pg(conn):
+        op.create_table('tool_mcp_providers',
+        sa.Column('id', models.types.StringUUID(), server_default=sa.text('uuid_generate_v4()'), nullable=False),
+        sa.Column('name', sa.String(length=40), nullable=False),
+        sa.Column('server_identifier', sa.String(length=24), nullable=False),
+        sa.Column('server_url', sa.Text(), nullable=False),
+        sa.Column('server_url_hash', sa.String(length=64), nullable=False),
+        sa.Column('icon', sa.String(length=255), nullable=True),
+        sa.Column('tenant_id', models.types.StringUUID(), nullable=False),
+        sa.Column('user_id', models.types.StringUUID(), nullable=False),
+        sa.Column('encrypted_credentials', sa.Text(), nullable=True),
+        sa.Column('authed', sa.Boolean(), nullable=False),
+        sa.Column('tools', sa.Text(), nullable=False),
+        sa.Column('created_at', sa.DateTime(), server_default=sa.text('CURRENT_TIMESTAMP(0)'), nullable=False),
+        sa.Column('updated_at', sa.DateTime(), server_default=sa.text('CURRENT_TIMESTAMP(0)'), nullable=False),
+        sa.PrimaryKeyConstraint('id', name='tool_mcp_provider_pkey'),
+        sa.UniqueConstraint('tenant_id', 'name', name='unique_mcp_provider_name'),
+        sa.UniqueConstraint('tenant_id', 'server_identifier', name='unique_mcp_provider_server_identifier'),
+        sa.UniqueConstraint('tenant_id', 'server_url_hash', name='unique_mcp_provider_server_url')
+        )
+    else:
+        op.create_table('tool_mcp_providers',
+        sa.Column('id', models.types.StringUUID(), default=lambda: str(uuid4()), nullable=False),
+        sa.Column('name', sa.String(length=40), nullable=False),
+        sa.Column('server_identifier', sa.String(length=24), nullable=False),
+        sa.Column('server_url', models.types.LongText(), nullable=False),
+        sa.Column('server_url_hash', sa.String(length=64), nullable=False),
+        sa.Column('icon', sa.String(length=255), nullable=True),
+        sa.Column('tenant_id', models.types.StringUUID(), nullable=False),
+        sa.Column('user_id', models.types.StringUUID(), nullable=False),
+        sa.Column('encrypted_credentials', models.types.LongText(), nullable=True),
+        sa.Column('authed', sa.Boolean(), nullable=False),
+        sa.Column('tools', models.types.LongText(), nullable=False),
+        sa.Column('created_at', sa.DateTime(), server_default=sa.func.current_timestamp(), nullable=False),
+        sa.Column('updated_at', sa.DateTime(), server_default=sa.func.current_timestamp(), nullable=False),
+        sa.PrimaryKeyConstraint('id', name='tool_mcp_provider_pkey'),
+        sa.UniqueConstraint('tenant_id', 'name', name='unique_mcp_provider_name'),
+        sa.UniqueConstraint('tenant_id', 'server_identifier', name='unique_mcp_provider_server_identifier'),
+        sa.UniqueConstraint('tenant_id', 'server_url_hash', name='unique_mcp_provider_server_url')
+        )
 
     # ### end Alembic commands ###
 

api/migrations/versions/2025_07_02_2332-1c9ba48be8e4_add_uuidv7_function_in_sql.py

@@ -27,6 +27,10 @@ import models as models
 import sqlalchemy as sa
 
 
+def _is_pg(conn):
+    return conn.dialect.name == "postgresql"
+
+
 # revision identifiers, used by Alembic.
 revision = '1c9ba48be8e4'
 down_revision = '58eb7bdb93fe'
@@ -40,7 +44,11 @@ def upgrade():
     # The ability to specify source timestamp has been removed because its type signature is incompatible with
     # PostgreSQL 18's `uuidv7` function. This capability is rarely needed in practice, as IDs can be
     # generated and controlled within the application layer.
-    op.execute(sa.text(r"""
+    conn = op.get_bind()
+    
+    if _is_pg(conn):
+        # PostgreSQL: Create uuidv7 functions
+        op.execute(sa.text(r"""
 /* Main function to generate a uuidv7 value with millisecond precision */
 CREATE FUNCTION uuidv7() RETURNS uuid
 AS
@@ -63,7 +71,7 @@ COMMENT ON FUNCTION uuidv7 IS
     'Generate a uuid-v7 value with a 48-bit timestamp (millisecond precision) and 74 bits of randomness';
 """))
 
-    op.execute(sa.text(r"""
+        op.execute(sa.text(r"""
 CREATE FUNCTION uuidv7_boundary(timestamptz) RETURNS uuid
 AS
 $$
@@ -79,8 +87,15 @@ COMMENT ON FUNCTION uuidv7_boundary(timestamptz) IS
     'Generate a non-random uuidv7 with the given timestamp (first 48 bits) and all random bits to 0. As the smallest possible uuidv7 for that timestamp, it may be used as a boundary for partitions.';
 """
 ))
+    else:
+        pass
 
 
 def downgrade():
-    op.execute(sa.text("DROP FUNCTION uuidv7"))
-    op.execute(sa.text("DROP FUNCTION uuidv7_boundary"))
+    conn = op.get_bind()
+    
+    if _is_pg(conn):
+        op.execute(sa.text("DROP FUNCTION uuidv7"))
+        op.execute(sa.text("DROP FUNCTION uuidv7_boundary"))
+    else:
+        pass