diff --git a/api/.importlinter b/api/.importlinter index 760c8d7c44b442..d53f0c2e1c9d74 100644 --- a/api/.importlinter +++ b/api/.importlinter @@ -1149,6 +1149,36 @@ forbidden_modules = # so only direct imports are checked here. allow_indirect_imports = True +[importlinter:contract:app-trigger-store-boundary] +name = App trigger management contracts are framework and persistence neutral +type = forbidden +source_modules = + services.trigger.app_trigger_store + services.trigger.app_trigger_service +forbidden_modules = + configs + controllers + extensions + flask + models + repositories + sqlalchemy + werkzeug + +[importlinter:contract:console-app-trigger-controller-boundary] +name = Console app trigger controller reaches persistence only through the composition root +type = forbidden +source_modules = + controllers.console.app.workflow_trigger +forbidden_modules = + extensions.ext_database + models.model + models.trigger + repositories + sqlalchemy +# The composition root wires repositories, so only direct imports are checked here. +allow_indirect_imports = True + [importlinter:contract:app-audio-boundary] name = App audio operations and values are framework and persistence neutral type = forbidden diff --git a/api/controllers/console/app/error.py b/api/controllers/console/app/error.py index adb0fbcf366a2a..f9812253a2eb8c 100644 --- a/api/controllers/console/app/error.py +++ b/api/controllers/console/app/error.py @@ -1,3 +1,5 @@ +from http import HTTPStatus + from libs.exception import BaseHTTPException @@ -7,6 +9,18 @@ class AppNotFoundError(BaseHTTPException): code = 404 +class AppTriggerNotFoundError(BaseHTTPException): + error_code = "app_trigger_not_found" + description = "Trigger not found." + code = HTTPStatus.NOT_FOUND + + +class WebhookTriggerNotFoundError(BaseHTTPException): + error_code = "webhook_trigger_not_found" + description = "Webhook trigger not found for this node." + code = HTTPStatus.NOT_FOUND + + class ProviderNotInitializeError(BaseHTTPException): error_code = "provider_not_initialize" description = ( diff --git a/api/controllers/console/app/workflow_trigger.py b/api/controllers/console/app/workflow_trigger.py index ab684a6d8f0925..45dc76ffcd8fd1 100644 --- a/api/controllers/console/app/workflow_trigger.py +++ b/api/controllers/console/app/workflow_trigger.py @@ -1,35 +1,36 @@ -import logging +from dataclasses import asdict from datetime import datetime +from http import HTTPStatus +from uuid import UUID from flask_restx import Resource from pydantic import BaseModel, field_validator -from sqlalchemy import select -from sqlalchemy.orm import Session -from werkzeug.exceptions import NotFound from configs import dify_config from controllers.common.rbac import PlainApp, RBACCheck -from controllers.common.schema import query_params_from_model, register_schema_models -from controllers.common.session import with_session +from controllers.common.schema import query_params_from_model, register_response_schema_models, register_schema_models from controllers.console import console_ns -from controllers.console.app.wraps import get_app_model -from controllers.console.wraps import ( - RBACPermission, - account_initialization_required, - edit_permission_required, - model_validate, - rbac_permission_required, - setup_required, - with_current_tenant_id, -) +from controllers.console.app.error import AppNotFoundError, AppTriggerNotFoundError, WebhookTriggerNotFoundError +from controllers.console.flask_admission import console_account_admission +from controllers.console.wraps import RBACPermission, model_validate +from core.trigger.utils.endpoint import generate_webhook_trigger_endpoint +from enums.account import TenantAccountRole +from extensions.ext_application_services import application_services from fields.base import ResponseModel +from libs.external_api import error_responses from libs.helper import dump_response -from libs.login import login_required -from models.enums import AppTriggerStatus -from models.model import App, AppMode -from models.trigger import AppTrigger, WorkflowWebhookTrigger - -logger = logging.getLogger(__name__) +from machinery.context import RequestContext +from models.enums import AppTriggerType +from services.app.console_service import ConsoleAppNotFoundError +from services.trigger.app_trigger_store import ( + AppTriggerNotFoundError as AppTriggerMissing, +) +from services.trigger.app_trigger_store import ( + AppTriggerRecord, +) +from services.trigger.app_trigger_store import ( + WebhookTriggerNotFoundError as WebhookTriggerMissing, +) class WebhookTriggerQuery(BaseModel): @@ -80,126 +81,86 @@ def _normalize_string_fields(cls, value: object) -> str: return str(value) -register_schema_models( +register_schema_models(console_ns, WebhookTriggerQuery, ParserEnable) +register_response_schema_models( console_ns, - WebhookTriggerQuery, - ParserEnable, WorkflowTriggerResponse, WorkflowTriggerListResponse, WebhookTriggerResponse, ) +def _with_trigger_icon(trigger: AppTriggerRecord) -> dict[str, object]: + icon = "" + if trigger.trigger_type == AppTriggerType.TRIGGER_PLUGIN: + icon = ( + dify_config.CONSOLE_API_URL + + "/console/api/workspaces/current/tool-provider/builtin/" + + (trigger.provider_name or "") + + "/icon" + ) + return {**asdict(trigger), "icon": icon} + + @console_ns.route("/apps//workflows/triggers/webhook") class WebhookTriggerApi(Resource): - """Webhook Trigger API""" - + @error_responses(AppNotFoundError, WebhookTriggerNotFoundError) @console_ns.doc(params=query_params_from_model(WebhookTriggerQuery)) - @setup_required - @login_required - @account_initialization_required - @console_ns.response(200, "Success", console_ns.models[WebhookTriggerResponse.__name__]) - @rbac_permission_required(RBACCheck(RBACPermission.APP_VIEW_LAYOUT, PlainApp())) - @with_session(write=False) - @get_app_model(mode=AppMode.WORKFLOW) + @console_ns.response(HTTPStatus.OK, "Success", console_ns.models[WebhookTriggerResponse.__name__]) + @console_account_admission(rbac_checks=[RBACCheck(RBACPermission.APP_VIEW_LAYOUT, PlainApp())]) @model_validate(WebhookTriggerQuery) - def get(self, req_data: WebhookTriggerQuery, session: Session, app_model: App): + def get(self, req_data: WebhookTriggerQuery, context: RequestContext, app_id: UUID) -> dict[str, object]: """Get webhook trigger for a node""" - - node_id = req_data.node_id - - # Get webhook trigger for this app and node - webhook_trigger = session.scalar( - select(WorkflowWebhookTrigger) - .where( - WorkflowWebhookTrigger.app_id == app_model.id, - WorkflowWebhookTrigger.node_id == node_id, - ) - .limit(1) + try: + webhook = application_services().apps.triggers.get_webhook(context, str(app_id), req_data.node_id) + except ConsoleAppNotFoundError as error: + raise AppNotFoundError(str(error)) from error + except WebhookTriggerMissing as error: + raise WebhookTriggerNotFoundError(str(error)) from error + return dump_response( + WebhookTriggerResponse, + { + **asdict(webhook), + "webhook_url": generate_webhook_trigger_endpoint(webhook.webhook_id), + "webhook_debug_url": generate_webhook_trigger_endpoint(webhook.webhook_id, debug=True), + }, ) - if not webhook_trigger: - raise NotFound("Webhook trigger not found for this node") - - return dump_response(WebhookTriggerResponse, webhook_trigger) - @console_ns.route("/apps//triggers") class AppTriggersApi(Resource): - """App Triggers list API""" - - @setup_required - @login_required - @account_initialization_required - @console_ns.response(200, "Success", console_ns.models[WorkflowTriggerListResponse.__name__]) - @with_current_tenant_id - @rbac_permission_required(RBACCheck(RBACPermission.APP_VIEW_LAYOUT, PlainApp())) - @with_session(write=False) - @get_app_model(mode=AppMode.WORKFLOW) - def get(self, session: Session, current_tenant_id: str, app_model: App): + @error_responses(AppNotFoundError) + @console_ns.response(HTTPStatus.OK, "Success", console_ns.models[WorkflowTriggerListResponse.__name__]) + @console_account_admission(rbac_checks=[RBACCheck(RBACPermission.APP_VIEW_LAYOUT, PlainApp())]) + def get(self, context: RequestContext, app_id: UUID) -> dict[str, object]: """Get app triggers list""" - # Get all triggers for this app using select API - triggers = ( - session.execute( - select(AppTrigger) - .where( - AppTrigger.tenant_id == current_tenant_id, - AppTrigger.app_id == app_model.id, - ) - .order_by(AppTrigger.created_at.desc(), AppTrigger.id.desc()) - ) - .scalars() - .all() + try: + triggers = application_services().apps.triggers.list_triggers(context, str(app_id)) + except ConsoleAppNotFoundError as error: + raise AppNotFoundError(str(error)) from error + return dump_response( + WorkflowTriggerListResponse, {"data": [_with_trigger_icon(trigger) for trigger in triggers]} ) - # Add computed icon field for each trigger - url_prefix = dify_config.CONSOLE_API_URL + "/console/api/workspaces/current/tool-provider/builtin/" - for trigger in triggers: - if trigger.trigger_type == "trigger-plugin": - trigger.icon = url_prefix + trigger.provider_name + "/icon" # type: ignore - else: - trigger.icon = "" # type: ignore - - return dump_response(WorkflowTriggerListResponse, {"data": triggers}) - @console_ns.route("/apps//trigger-enable") class AppTriggerEnableApi(Resource): + @error_responses(AppNotFoundError, AppTriggerNotFoundError) @console_ns.expect(console_ns.models[ParserEnable.__name__]) - @setup_required - @login_required - @account_initialization_required - @edit_permission_required - @rbac_permission_required(RBACCheck(RBACPermission.APP_EDIT, PlainApp())) - @console_ns.response(200, "Success", console_ns.models[WorkflowTriggerResponse.__name__]) - @with_current_tenant_id - @with_session - @get_app_model(mode=AppMode.WORKFLOW) + @console_ns.response(HTTPStatus.OK, "Success", console_ns.models[WorkflowTriggerResponse.__name__]) + @console_account_admission( + allowed_roles=frozenset({TenantAccountRole.OWNER, TenantAccountRole.ADMIN, TenantAccountRole.EDITOR}), + rbac_checks=[RBACCheck(RBACPermission.APP_EDIT, PlainApp())], + ) @model_validate(ParserEnable) - def post(self, req_data: ParserEnable, session: Session, current_tenant_id: str, app_model: App): + def post(self, req_data: ParserEnable, context: RequestContext, app_id: UUID) -> dict[str, object]: """Update app trigger (enable/disable)""" - - trigger_id = req_data.trigger_id - # Find the trigger using select - trigger = session.execute( - select(AppTrigger).where( - AppTrigger.id == trigger_id, - AppTrigger.tenant_id == current_tenant_id, - AppTrigger.app_id == app_model.id, + try: + trigger = application_services().apps.triggers.set_enabled( + context, str(app_id), req_data.trigger_id, enabled=req_data.enable_trigger ) - ).scalar_one_or_none() - - if not trigger: - raise NotFound("Trigger not found") - - # Update status based on enable_trigger boolean - trigger.status = AppTriggerStatus.ENABLED if req_data.enable_trigger else AppTriggerStatus.DISABLED - - # Add computed icon field - url_prefix = dify_config.CONSOLE_API_URL + "/console/api/workspaces/current/tool-provider/builtin/" - if trigger.trigger_type == "trigger-plugin": - trigger.icon = url_prefix + trigger.provider_name + "/icon" # type: ignore - else: - trigger.icon = "" # type: ignore - - return dump_response(WorkflowTriggerResponse, trigger) + except ConsoleAppNotFoundError as error: + raise AppNotFoundError(str(error)) from error + except AppTriggerMissing as error: + raise AppTriggerNotFoundError(str(error)) from error + return dump_response(WorkflowTriggerResponse, _with_trigger_icon(trigger)) diff --git a/api/controllers/trigger/webhook.py b/api/controllers/trigger/webhook.py index b9734cb9f39db7..a2f46418de2eaa 100644 --- a/api/controllers/trigger/webhook.py +++ b/api/controllers/trigger/webhook.py @@ -77,6 +77,7 @@ def handle_webhook(webhook_id: str): webhook_data, workflow, end_users=application_services().app_scoped_end_users.commands, + trigger_status=application_services().apps.trigger_status, ) # Return configured response diff --git a/api/events/event_handlers/update_app_triggers_when_app_published_workflow_updated.py b/api/events/event_handlers/update_app_triggers_when_app_published_workflow_updated.py index d55fe262fb25a3..566c1fba71fff6 100644 --- a/api/events/event_handlers/update_app_triggers_when_app_published_workflow_updated.py +++ b/api/events/event_handlers/update_app_triggers_when_app_published_workflow_updated.py @@ -1,112 +1,20 @@ -from typing import cast +"""Register published trigger nodes in the publisher's transaction.""" -from sqlalchemy import select -from sqlalchemy.orm import sessionmaker +from sqlalchemy.orm import Session -from core.trigger.constants import TRIGGER_NODE_TYPES from events.app_event import app_published_workflow_was_updated -from extensions.ext_database import db -from models import AppMode -from models.enums import AppTriggerStatus -from models.trigger import AppTrigger +from extensions.ext_application_services import application_services +from models.model import App, AppMode from models.workflow import Workflow @app_published_workflow_was_updated.connect -def handle(sender, **kwargs): - """ - Handle app published workflow update event to sync app_triggers table. - - When a workflow is published, this handler will: - 1. Extract trigger nodes from the workflow graph - 2. Compare with existing app_triggers records - 3. Add new triggers and remove obsolete ones - """ - app = sender - if app.mode != AppMode.WORKFLOW.value: +def handle(sender: App, *, session: Session, published_workflow: Workflow | None = None, **kwargs: object) -> None: + if sender.mode != AppMode.WORKFLOW or published_workflow is None: return - - published_workflow = kwargs.get("published_workflow") - published_workflow = cast(Workflow, published_workflow) - # Extract trigger info from workflow - trigger_infos = get_trigger_infos_from_workflow(published_workflow) - - with sessionmaker(db.engine).begin() as session: - # Get existing app triggers - existing_triggers = ( - session.execute( - select(AppTrigger).where(AppTrigger.tenant_id == app.tenant_id, AppTrigger.app_id == app.id) - ) - .scalars() - .all() - ) - - # Convert existing triggers to dict for easy lookup - existing_triggers_map = {trigger.node_id: trigger for trigger in existing_triggers} - - # Get current and new node IDs - existing_node_ids = set(existing_triggers_map.keys()) - new_node_ids = {info["node_id"] for info in trigger_infos} - - # Calculate changes - added_node_ids = new_node_ids - existing_node_ids - removed_node_ids = existing_node_ids - new_node_ids - - # Remove obsolete triggers - for node_id in removed_node_ids: - session.delete(existing_triggers_map[node_id]) - - for trigger_info in trigger_infos: - node_id = trigger_info["node_id"] - - if node_id in added_node_ids: - # Create new trigger - app_trigger = AppTrigger( - tenant_id=app.tenant_id, - app_id=app.id, - trigger_type=trigger_info["node_type"], - title=trigger_info["node_title"], - node_id=node_id, - provider_name=trigger_info.get("node_provider_name", ""), - status=AppTriggerStatus.ENABLED, - ) - session.add(app_trigger) - elif node_id in existing_node_ids: - # Update existing trigger if needed - existing_trigger = existing_triggers_map[node_id] - new_title = trigger_info["node_title"] - if new_title and existing_trigger.title != new_title: - existing_trigger.title = new_title - session.add(existing_trigger) - - -def get_trigger_infos_from_workflow(published_workflow: Workflow) -> list[dict]: - """ - Extract trigger node information from the workflow graph. - - Returns: - List of trigger info dictionaries containing: - - node_type: The type of the trigger node ('trigger-webhook', 'trigger-schedule', 'trigger-plugin') - - node_id: The node ID in the workflow - - node_title: The title of the node - - node_provider_name: The name of the node's provider, only for plugin - """ graph = published_workflow.graph_dict - if not graph: - return [] - - nodes = graph.get("nodes", []) - trigger_types = TRIGGER_NODE_TYPES - - trigger_infos = [ - { - "node_type": node.get("data", {}).get("type"), - "node_id": node.get("id"), - "node_title": node.get("data", {}).get("title"), - "node_provider_name": node.get("data", {}).get("provider_name"), - } - for node in nodes - if node.get("data", {}).get("type") in trigger_types - ] - - return trigger_infos + application_services().apps.create_trigger_publication_service(session).sync( + tenant_id=sender.tenant_id, + app_id=sender.id, + nodes=graph.get("nodes", []), + ) diff --git a/api/extensions/application_services/app.py b/api/extensions/application_services/app.py index 8459402af5040b..cd29847a8c4c5a 100644 --- a/api/extensions/application_services/app.py +++ b/api/extensions/application_services/app.py @@ -1,6 +1,8 @@ """Composition of App use cases.""" +from collections.abc import Callable from dataclasses import dataclass +from functools import partial from sqlalchemy.orm import Session, sessionmaker @@ -13,6 +15,7 @@ from repositories.app.enterprise_app_repository import EnterpriseAppRepository from repositories.app.export_repository import AppExportRepository from repositories.factory import DifyAPIRepositoryFactory +from repositories.trigger.app_repository import SQLAlchemyAppTriggerRepository from repositories.workspace.workspace_repository import WorkspaceRepository from services.agent.roster_package_exporter import RosterAgentPackageExporter from services.agent.roster_package_importer import RosterAgentPackageImporter @@ -32,6 +35,8 @@ from services.oauth_server_service import OAuthServerService from services.rbac.members import MemberService from services.recommended_app_package_service import RecommendedAppPackageService +from services.trigger.app_trigger_service import AppTriggerPublicationService, AppTriggerService +from services.trigger.app_trigger_store import AppTriggerStore from services.workflow.dsl_import import DslImportService @@ -43,6 +48,9 @@ class AppServices: discovery: AppDiscoveryService console: ConsoleAppService enterprise: EnterpriseAppService + triggers: AppTriggerStore + trigger_status: AppTriggerService + create_trigger_publication_service: Callable[[Session], AppTriggerPublicationService] def build_app_services( @@ -61,6 +69,7 @@ def build_app_services( agent_importer=RosterAgentPackageImporter(), recommended_packages=recommended_packages, ) + triggers = SQLAlchemyAppTriggerRepository(session_factory=database_client) return AppServices( imports=AppImportService(accounts=SQLAlchemyAccountRepository(database_client), definitions=transfers), exports=AppExportService( @@ -89,6 +98,12 @@ def build_app_services( workflow_runs=DifyAPIRepositoryFactory.create_api_workflow_run_repository(session_maker=database_client), now=naive_utc_now, ), + triggers=triggers, + trigger_status=AppTriggerService(triggers=triggers), + # Retain the session only within the caller's publication transaction. + create_trigger_publication_service=lambda session: AppTriggerPublicationService( + sync_triggers=partial(triggers.sync_publication, session=session) + ), ) diff --git a/api/openapi/markdown/console-openapi.md b/api/openapi/markdown/console-openapi.md index 80f46c998c748a..5df10a6814188f 100644 --- a/api/openapi/markdown/console-openapi.md +++ b/api/openapi/markdown/console-openapi.md @@ -3735,7 +3735,7 @@ Create a new tracing configuration for an application | Code | Description | Schema | | ---- | ----------- | ------ | | 200 | Success | **application/json**: [WorkflowTriggerResponse](#workflowtriggerresponse)
| -| 404 | Not Found | **application/json**: [AppNotFoundError](#appnotfounderror)
[ErrorResponse](#errorresponse)
| +| 404 | Not Found | **application/json**: [AppNotFoundError](#appnotfounderror)
[AppTriggerNotFoundError](#apptriggernotfounderror)
[ErrorResponse](#errorresponse)
| | default | Error | **application/json**: [ErrorResponse](#errorresponse)
| ### [GET] /apps/{app_id}/triggers @@ -5276,7 +5276,7 @@ Full value for one declared output of a published run. | Code | Description | Schema | | ---- | ----------- | ------ | | 200 | Success | **application/json**: [WebhookTriggerResponse](#webhooktriggerresponse)
| -| 404 | Not Found | **application/json**: [AppNotFoundError](#appnotfounderror)
[ErrorResponse](#errorresponse)
| +| 404 | Not Found | **application/json**: [AppNotFoundError](#appnotfounderror)
[WebhookTriggerNotFoundError](#webhooktriggernotfounderror)
[ErrorResponse](#errorresponse)
| | default | Error | **application/json**: [ErrorResponse](#errorresponse)
| ### [DELETE] /apps/{app_id}/workflows/{workflow_id} @@ -18933,6 +18933,14 @@ Function Suggested questions after answer disabled. | enabled | boolean | | No | | tracing_provider | string | | No | +#### AppTriggerNotFoundError + +Trigger not found. + +| Name | Type | Description | Required | +| ---- | ---- | ----------- | -------- | +| AppTriggerNotFoundError | | Trigger not found. | | + #### AppUnavailableError App unavailable, please check your app configurations. @@ -28617,6 +28625,14 @@ The app access service is unavailable. Try again later. | ---- | ---- | ----------- | -------- | | protocol | [SSOProtocol](#ssoprotocol) | | Yes | +#### WebhookTriggerNotFoundError + +Webhook trigger not found for this node. + +| Name | Type | Description | Required | +| ---- | ---- | ----------- | -------- | +| WebhookTriggerNotFoundError | | Webhook trigger not found for this node. | | + #### WebhookTriggerQuery | Name | Type | Description | Required | diff --git a/api/repositories/trigger/app_repository.py b/api/repositories/trigger/app_repository.py new file mode 100644 index 00000000000000..1895bb0fd8090a --- /dev/null +++ b/api/repositories/trigger/app_repository.py @@ -0,0 +1,139 @@ +"""App trigger management with workspace-scoped, bounded database transactions.""" + +from collections.abc import Sequence +from typing import override + +from sqlalchemy import select, update +from sqlalchemy.orm import Session, sessionmaker + +from machinery.context import RequestContext +from models.enums import AppTriggerStatus +from models.model import AppMode +from models.trigger import AppTrigger, WorkflowWebhookTrigger +from repositories.app.console_repository import require_console_app +from services.app.console_service import ConsoleAppNotFoundError +from services.trigger.app_trigger_store import ( + AppTriggerNotFoundError, + AppTriggerRecord, + AppTriggerStore, + PublishedAppTrigger, + WebhookTriggerNotFoundError, + WebhookTriggerRecord, +) + + +def _require_workflow_app(session: Session, context: RequestContext, app_id: str) -> None: + app = require_console_app(session, context, app_id) + if app.mode != AppMode.WORKFLOW: + raise ConsoleAppNotFoundError(f"App mode is not in the supported list: {{{AppMode.WORKFLOW.value!r}}}") + + +def _trigger_record(trigger: AppTrigger) -> AppTriggerRecord: + return AppTriggerRecord( + id=trigger.id, + trigger_type=trigger.trigger_type, + title=trigger.title, + node_id=trigger.node_id, + provider_name=trigger.provider_name, + status=trigger.status, + created_at=trigger.created_at, + updated_at=trigger.updated_at, + ) + + +class SQLAlchemyAppTriggerRepository(AppTriggerStore): + def __init__(self, *, session_factory: sessionmaker[Session]) -> None: + self._session_factory = session_factory + + @override + def get_webhook(self, context: RequestContext, app_id: str, node_id: str) -> WebhookTriggerRecord: + with self._session_factory() as session: + _require_workflow_app(session, context, app_id) + webhook = session.scalar( + select(WorkflowWebhookTrigger).where( + WorkflowWebhookTrigger.tenant_id == context.active_workspace_id, + WorkflowWebhookTrigger.app_id == app_id, + WorkflowWebhookTrigger.node_id == node_id, + ) + ) + if webhook is None: + raise WebhookTriggerNotFoundError("Webhook trigger not found for this node") + return WebhookTriggerRecord( + id=webhook.id, + webhook_id=webhook.webhook_id, + node_id=webhook.node_id, + created_at=webhook.created_at, + ) + + @override + def list_triggers(self, context: RequestContext, app_id: str) -> tuple[AppTriggerRecord, ...]: + with self._session_factory() as session: + _require_workflow_app(session, context, app_id) + triggers = session.scalars( + select(AppTrigger) + .where(AppTrigger.tenant_id == context.active_workspace_id, AppTrigger.app_id == app_id) + .order_by(AppTrigger.created_at.desc(), AppTrigger.id.desc()) + ) + return tuple(_trigger_record(trigger) for trigger in triggers) + + @override + def set_enabled(self, context: RequestContext, app_id: str, trigger_id: str, *, enabled: bool) -> AppTriggerRecord: + with self._session_factory.begin() as session: + _require_workflow_app(session, context, app_id) + trigger = session.scalar( + select(AppTrigger).where( + AppTrigger.id == trigger_id, + AppTrigger.tenant_id == context.active_workspace_id, + AppTrigger.app_id == app_id, + ) + ) + if trigger is None: + raise AppTriggerNotFoundError("Trigger not found") + trigger.status = AppTriggerStatus.ENABLED if enabled else AppTriggerStatus.DISABLED + session.flush() + return _trigger_record(trigger) + + def sync_publication( + self, *, session: Session, tenant_id: str, app_id: str, triggers: Sequence[PublishedAppTrigger] + ) -> None: + """Join the publication transaction; registration must roll back with its graph. + + WorkflowService currently owns the publication Session and its version-counter + lock. Do not commit or create a second transaction here. Remove this borrowed + Session once publication exposes a unit of work for all graph registrations. + """ + existing = { + trigger.node_id: trigger + for trigger in session.scalars( + select(AppTrigger).where(AppTrigger.tenant_id == tenant_id, AppTrigger.app_id == app_id) + ) + } + current_node_ids = {trigger.node_id for trigger in triggers} + for node_id, trigger in existing.items(): + if node_id not in current_node_ids: + session.delete(trigger) + for trigger in triggers: + stored = existing.get(trigger.node_id) + if stored is None: + session.add( + AppTrigger( + tenant_id=tenant_id, + app_id=app_id, + node_id=trigger.node_id, + trigger_type=trigger.trigger_type, + title=trigger.title, + provider_name=trigger.provider_name, + status=AppTriggerStatus.ENABLED, + ) + ) + elif trigger.title and trigger.title != stored.title: + # Preserve status and provider/type metadata for existing node IDs. + stored.title = trigger.title + + def mark_tenant_triggers_rate_limited(self, tenant_id: str) -> None: + with self._session_factory.begin() as session: + session.execute( + update(AppTrigger) + .where(AppTrigger.tenant_id == tenant_id, AppTrigger.status == AppTriggerStatus.ENABLED) + .values(status=AppTriggerStatus.RATE_LIMITED) + ) diff --git a/api/services/trigger/app_trigger_service.py b/api/services/trigger/app_trigger_service.py index 723d29e947feff..af840f75620290 100644 --- a/api/services/trigger/app_trigger_service.py +++ b/api/services/trigger/app_trigger_service.py @@ -1,45 +1,48 @@ -""" -AppTrigger management service. - -Handles AppTrigger model CRUD operations and status management. -This service centralizes all AppTrigger-related business logic. -""" +"""Publication policy and best-effort quota status updates for application triggers.""" import logging +from collections.abc import Iterable, Mapping +from typing import Any -from sqlalchemy import update -from sqlalchemy.orm import sessionmaker - -from extensions.ext_database import db -from models.enums import AppTriggerStatus -from models.trigger import AppTrigger +from core.trigger.constants import TRIGGER_NODE_TYPES +from services.trigger.app_trigger_store import ( + AppTriggerPublicationWriter, + AppTriggerStatusWriter, + PublishedAppTrigger, +) logger = logging.getLogger(__name__) class AppTriggerService: - """Service for managing AppTrigger lifecycle and status.""" - - @staticmethod - def mark_tenant_triggers_rate_limited(tenant_id: str) -> None: - """ - Mark all enabled triggers for a tenant as rate limited due to quota exceeded. + """Keep status persistence failures from replacing the original quota rejection.""" - This method is called when a tenant's quota is exhausted. It updates all - enabled triggers to RATE_LIMITED status to prevent further executions until - quota is restored. + def __init__(self, *, triggers: AppTriggerStatusWriter) -> None: + self._triggers = triggers - Args: - tenant_id: Tenant ID whose triggers should be marked as rate limited - - """ + def mark_tenant_triggers_rate_limited(self, tenant_id: str) -> None: try: - with sessionmaker(bind=db.engine).begin() as session: - session.execute( - update(AppTrigger) - .where(AppTrigger.tenant_id == tenant_id, AppTrigger.status == AppTriggerStatus.ENABLED) - .values(status=AppTriggerStatus.RATE_LIMITED) - ) - logger.info("Marked all enabled triggers as rate limited for tenant %s", tenant_id) + self._triggers.mark_tenant_triggers_rate_limited(tenant_id) + logger.info("Marked all enabled triggers as rate limited for tenant %s", tenant_id) except Exception: logger.exception("Failed to mark all enabled triggers as rate limited for tenant %s", tenant_id) + + +class AppTriggerPublicationService: + """Derive published trigger registration without changing existing enablement.""" + + def __init__(self, *, sync_triggers: AppTriggerPublicationWriter) -> None: + self._sync_triggers = sync_triggers + + def sync(self, *, tenant_id: str, app_id: str, nodes: Iterable[Mapping[str, Any]]) -> None: + triggers = tuple( + PublishedAppTrigger( + node_id=node["id"], + trigger_type=node["data"]["type"], + title=node["data"].get("title"), + provider_name=node["data"].get("provider_name"), + ) + for node in nodes + if node.get("data", {}).get("type") in TRIGGER_NODE_TYPES + ) + self._sync_triggers(tenant_id=tenant_id, app_id=app_id, triggers=triggers) diff --git a/api/services/trigger/app_trigger_store.py b/api/services/trigger/app_trigger_store.py new file mode 100644 index 00000000000000..3e680ad9bb3abf --- /dev/null +++ b/api/services/trigger/app_trigger_store.py @@ -0,0 +1,66 @@ +"""Detached records and persistence contract for Console app trigger management.""" + +from collections.abc import Sequence +from dataclasses import dataclass +from datetime import datetime +from typing import Protocol + +from machinery.context import RequestContext + + +class WebhookTriggerNotFoundError(Exception): + """The requested node has no webhook in the admitted app and workspace.""" + + +class AppTriggerNotFoundError(Exception): + """The trigger does not belong to the admitted app and workspace.""" + + +@dataclass(frozen=True, slots=True) +class AppTriggerRecord: + """Persisted values; None preserves a missing legacy node or provider name.""" + + id: str + trigger_type: str + title: str + node_id: str | None + provider_name: str | None + status: str + created_at: datetime + updated_at: datetime + + +@dataclass(frozen=True, slots=True) +class WebhookTriggerRecord: + id: str + webhook_id: str + node_id: str + created_at: datetime + + +class AppTriggerStore(Protocol): + def get_webhook(self, context: RequestContext, app_id: str, node_id: str) -> WebhookTriggerRecord: ... + + def list_triggers(self, context: RequestContext, app_id: str) -> tuple[AppTriggerRecord, ...]: ... + + def set_enabled( + self, context: RequestContext, app_id: str, trigger_id: str, *, enabled: bool + ) -> AppTriggerRecord: ... + + +@dataclass(frozen=True, slots=True) +class PublishedAppTrigger: + """Trigger identity and display metadata extracted from a published graph.""" + + node_id: str + trigger_type: str + title: str + provider_name: str | None + + +class AppTriggerPublicationWriter(Protocol): + def __call__(self, *, tenant_id: str, app_id: str, triggers: Sequence[PublishedAppTrigger]) -> None: ... + + +class AppTriggerStatusWriter(Protocol): + def mark_tenant_triggers_rate_limited(self, tenant_id: str) -> None: ... diff --git a/api/services/trigger/webhook_service.py b/api/services/trigger/webhook_service.py index df7ccc1af32e3b..71cf399b92ebeb 100644 --- a/api/services/trigger/webhook_service.py +++ b/api/services/trigger/webhook_service.py @@ -807,6 +807,7 @@ def trigger_workflow_execution( workflow: Workflow, *, end_users: WebhookEndUserProvisioner, + trigger_status: AppTriggerService, ) -> None: """Trigger workflow execution via AsyncWorkflowService. @@ -841,7 +842,7 @@ def trigger_workflow_execution( try: quota_charge = QuotaService.reserve(QuotaType.TRIGGER, webhook_trigger.tenant_id) except QuotaExceededError: - AppTriggerService.mark_tenant_triggers_rate_limited(webhook_trigger.tenant_id) + trigger_status.mark_tenant_triggers_rate_limited(webhook_trigger.tenant_id) raise try: diff --git a/api/tasks/trigger_processing_tasks.py b/api/tasks/trigger_processing_tasks.py index 7aa7bfb554e8ba..7acfad87e7e421 100644 --- a/api/tasks/trigger_processing_tasks.py +++ b/api/tasks/trigger_processing_tasks.py @@ -243,6 +243,7 @@ def dispatch_triggered_workflow( end_users: TriggerEndUserProvisioner, events: TriggerEventService, bindings: TriggerSubscriptionBindingQuery, + trigger_status: AppTriggerService, ) -> int: """Process triggered workflows. @@ -254,6 +255,7 @@ def dispatch_triggered_workflow( end_users: Provisioner that resolves or creates the end users for each app events: Shared plugin event preparation and invocation bindings: Workspace-scoped subscriber lookup + trigger_status: Best-effort quota status updates Returns: The number of workflows dispatched for the event @@ -316,7 +318,7 @@ def dispatch_triggered_workflow( try: quota_charge = QuotaService.reserve(QuotaType.TRIGGER, subscription.tenant_id) except QuotaExceededError: - AppTriggerService.mark_tenant_triggers_rate_limited(subscription.tenant_id) + trigger_status.mark_tenant_triggers_rate_limited(subscription.tenant_id) logger.info("Tenant %s rate limited, skipping plugin trigger %s", subscription.tenant_id, plugin_trigger.id) return dispatched_count @@ -414,6 +416,7 @@ def dispatch_triggered_workflows( end_users: TriggerEndUserProvisioner, event_service: TriggerEventService, bindings: TriggerSubscriptionBindingQuery, + trigger_status: AppTriggerService, ) -> int: dispatched_count = 0 for event_name in events: @@ -426,6 +429,7 @@ def dispatch_triggered_workflows( end_users=end_users, events=event_service, bindings=bindings, + trigger_status=trigger_status, ) except Exception: logger.exception( @@ -499,6 +503,7 @@ def dispatch_triggered_workflows_async( end_users=application_services().app_scoped_end_users.commands, event_service=application_services().workspaces.trigger_events, bindings=application_services().workspaces.trigger_binding_queries, + trigger_status=application_services().apps.trigger_status, ) debug_dispatched = dispatch_trigger_debug_event( diff --git a/api/tasks/workflow_schedule_tasks.py b/api/tasks/workflow_schedule_tasks.py index 36ed189c7c1cc8..bc8192a6eea035 100644 --- a/api/tasks/workflow_schedule_tasks.py +++ b/api/tasks/workflow_schedule_tasks.py @@ -9,11 +9,11 @@ TenantOwnerNotFoundError, ) from enums import QuotaType +from extensions.ext_application_services import application_services from models.trigger import WorkflowSchedulePlan from services.async_workflow_service import AsyncWorkflowService from services.errors.app import QuotaExceededError from services.quota_service import QuotaService, unlimited -from services.trigger.app_trigger_service import AppTriggerService from services.trigger.schedule_service import ScheduleService from services.workflow.entities import ScheduleTriggerData @@ -47,7 +47,7 @@ def run_schedule_trigger(schedule_id: str) -> None: try: quota_charge = QuotaService.reserve(QuotaType.TRIGGER, schedule.tenant_id) except QuotaExceededError: - AppTriggerService.mark_tenant_triggers_rate_limited(schedule.tenant_id) + application_services().apps.trigger_status.mark_tenant_triggers_rate_limited(schedule.tenant_id) logger.info("Tenant %s rate limited, skipping schedule trigger %s", schedule.tenant_id, schedule_id) return diff --git a/api/tests/test_containers_integration_tests/services/test_trigger_binding_publication.py b/api/tests/test_containers_integration_tests/services/test_trigger_binding_publication.py index 2aa25b8034b929..d6b3f5488f505d 100644 --- a/api/tests/test_containers_integration_tests/services/test_trigger_binding_publication.py +++ b/api/tests/test_containers_integration_tests/services/test_trigger_binding_publication.py @@ -136,3 +136,4 @@ def test_publish_signal_shares_workflow_and_binding_transaction( assert saved_binding.subscription_id == (next_subscription_id if commit else previous_subscription_id) assert saved_binding.event_name == ("issues" if commit else "push") assert (reader.get(Workflow, new_workflow_id) is not None) == commit + assert (reader.scalar(select(AppTrigger.id).where(AppTrigger.app_id == app_id)) is not None) == commit diff --git a/api/tests/test_containers_integration_tests/services/test_webhook_service_relationships.py b/api/tests/test_containers_integration_tests/services/test_webhook_service_relationships.py index acb786fc7585e7..c66cc5de118d59 100644 --- a/api/tests/test_containers_integration_tests/services/test_webhook_service_relationships.py +++ b/api/tests/test_containers_integration_tests/services/test_webhook_service_relationships.py @@ -13,6 +13,7 @@ from core.trigger.constants import TRIGGER_WEBHOOK_NODE_TYPE from enums import QuotaType +from extensions.ext_application_services import application_services from models.account import Account, Tenant, TenantAccountJoin, TenantAccountRole from models.enums import AppTriggerStatus, AppTriggerType, EndUserType from models.model import App, EndUser @@ -373,6 +374,7 @@ def test_trigger_workflow_execution_triggers_async_workflow_successfully( webhook_data, workflow, end_users=_EndUserServiceStub(end_user), + trigger_status=application_services().apps.trigger_status, ) mock_reserve.assert_called_once() @@ -401,14 +403,12 @@ def test_trigger_workflow_execution_marks_tenant_rate_limited_when_quota_exceede db_session_with_containers, app=app, account=account, node_id="node-1" ) - with ( - patch( - "services.trigger.webhook_service.QuotaService.reserve", - side_effect=QuotaExceededError(feature="trigger", tenant_id=tenant.id, required=1), - ), - patch( - "services.trigger.webhook_service.AppTriggerService.mark_tenant_triggers_rate_limited" - ) as mock_mark_rate_limited, + app_trigger = factory.create_app_trigger( + db_session_with_containers, app=app, node_id="node-1", status=AppTriggerStatus.ENABLED + ) + with patch( + "services.trigger.webhook_service.QuotaService.reserve", + side_effect=QuotaExceededError(feature="trigger", tenant_id=tenant.id, required=1), ): with pytest.raises(QuotaExceededError): WebhookService.trigger_workflow_execution( @@ -416,9 +416,11 @@ def test_trigger_workflow_execution_marks_tenant_rate_limited_when_quota_exceede {"body": {}, "headers": {}, "query_params": {}, "files": {}, "method": "POST"}, workflow, end_users=_EndUserServiceStub(_end_user(tenant_id=tenant.id, app_id=app.id)), + trigger_status=application_services().apps.trigger_status, ) - mock_mark_rate_limited.assert_called_once_with(tenant.id) + db_session_with_containers.refresh(app_trigger) + assert app_trigger.status == AppTriggerStatus.RATE_LIMITED def test_trigger_workflow_execution_logs_and_reraises_unexpected_errors( self, @@ -444,6 +446,7 @@ def test_trigger_workflow_execution_logs_and_reraises_unexpected_errors( {"body": {}, "headers": {}, "query_params": {}, "files": {}, "method": "POST"}, workflow, end_users=_EndUserServiceStub(RuntimeError("boom")), + trigger_status=application_services().apps.trigger_status, ) assert caplog.messages.count(f"Failed to trigger workflow for webhook {webhook_trigger.webhook_id}") == 1 diff --git a/api/tests/unit_tests/controllers/console/app/test_app_apis.py b/api/tests/unit_tests/controllers/console/app/test_app_apis.py index bdbd6282f7c0c7..b68a3fb717dced 100644 --- a/api/tests/unit_tests/controllers/console/app/test_app_apis.py +++ b/api/tests/unit_tests/controllers/console/app/test_app_apis.py @@ -45,9 +45,6 @@ from controllers.console.app import ( workflow_statistic as workflow_statistic_module, ) -from controllers.console.app import ( - workflow_trigger as workflow_trigger_module, -) from controllers.console.app import ( wraps as wraps_module, ) @@ -68,12 +65,10 @@ WorkflowDraftVariableUpdatePayload, ) from controllers.console.app.workflow_statistic import WorkflowStatisticQuery -from controllers.console.app.workflow_trigger import ParserEnable, WebhookTriggerQuery from machinery.context import RequestContext from models import App, Site from models.account import Account, AccountStatus from models.engine import db -from models.trigger import WorkflowWebhookTrigger from services.app_site_service import ( AppSiteAppNotFoundError, AppSiteChanges, @@ -130,7 +125,6 @@ def database_app() -> Iterator[Flask]: with app.app_context(): Site.__table__.create(db.engine) - WorkflowWebhookTrigger.__table__.create(db.engine) yield app @@ -782,39 +776,6 @@ def _patch_statistic_time_range(monkeypatch: pytest.MonkeyPatch) -> None: ) -class TestWorkflowTriggerEndpoints: - def test_webhook_trigger_payload(self): - payload = WebhookTriggerQuery(node_id="node-1") - assert payload.node_id == "node-1" - - enable_payload = ParserEnable(trigger_id="trigger-1", enable_trigger=True) - assert enable_payload.enable_trigger is True - - def test_webhook_trigger_api_get( - self, - database_app: Flask, - sqlite_session: Session, - ) -> None: - api = workflow_trigger_module.WebhookTriggerApi() - method = unwrap(api.get) - trigger = WorkflowWebhookTrigger( - app_id=APP_ID, - node_id="node-1", - tenant_id=TENANT_ID, - webhook_id="webhook-1", - created_by=USER_ID, - ) - sqlite_session.add(trigger) - sqlite_session.commit() - - with database_app.test_request_context("/?node_id=node-1"): - result = method(api, WebhookTriggerQuery(node_id="node-1"), sqlite_session, app_model=_make_app()) - - assert isinstance(result, dict) - assert {"id", "webhook_id", "webhook_url", "webhook_debug_url", "node_id", "created_at"} <= set(result.keys()) - assert result["webhook_id"] == "webhook-1" - - class TestWrapsEndpoints: def test_get_app_model_context(self): assert hasattr(wraps_module, "get_app_model") diff --git a/api/tests/unit_tests/controllers/console/app/test_workflow_trigger_api.py b/api/tests/unit_tests/controllers/console/app/test_workflow_trigger_api.py index 1d931b0be533e7..6a26681f11aea1 100644 --- a/api/tests/unit_tests/controllers/console/app/test_workflow_trigger_api.py +++ b/api/tests/unit_tests/controllers/console/app/test_workflow_trigger_api.py @@ -1,145 +1,459 @@ -from __future__ import annotations +"""Workflow trigger routes use real admission, composition, and SQLite transactions.""" -import inspect -import uuid -from datetime import UTC, datetime -from unittest.mock import PropertyMock, patch +from collections.abc import Iterator +from dataclasses import dataclass +from datetime import datetime +from http import HTTPStatus +from uuid import uuid4 -from flask import Flask -from sqlalchemy.orm import Session +import pytest +from flask import Blueprint, Flask, Response +from flask.testing import FlaskClient +from flask_restx import Resource +from redis import Redis +from sqlalchemy import Connection, Engine, event, select, text +from sqlalchemy.orm import Session, SessionTransaction, sessionmaker +from werkzeug.test import TestResponse -from controllers.console import console_ns -from controllers.console.app import workflow_trigger as workflow_trigger_module +from constants import HEADER_NAME_CSRF_TOKEN +from controllers.console.app import workflow_trigger +from controllers.console.wraps import _is_setup_completed +from core.trigger.utils.endpoint import base_url +from enums import DeploymentEdition +from extensions.ext_application_services import build_application_services +from extensions.ext_login import DifyLoginManager, bind_account_loader, load_user_from_request, unauthorized_handler +from extensions.ext_redis import RedisClientWrapper +from libs.datetime_utils import naive_utc_now +from libs.external_api import ExternalApi +from libs.passport import PassportService +from libs.token import _real_cookie_name, generate_csrf_token +from models import Account, DifySetup, Tenant, TenantAccountJoin +from models.account import AccountStatus, TenantAccountRole +from models.agent import Agent, AgentScope, AgentSource +from models.engine import db from models.enums import AppTriggerStatus, AppTriggerType -from models.model import App, AppMode, IconType -from models.trigger import AppTrigger - - -def _persist_app_trigger( - session: Session, - *, - tenant_id: str | None = None, - status: AppTriggerStatus = AppTriggerStatus.ENABLED, -) -> tuple[App, AppTrigger]: - tenant_id = tenant_id or str(uuid.uuid4()) - app_model = App( - id=str(uuid.uuid4()), - tenant_id=tenant_id, - name="Workflow App", - mode=AppMode.WORKFLOW, - icon_type=IconType.EMOJI, - icon="workflow", - icon_background="#FFFFFF", - enable_site=False, - enable_api=True, - ) - trigger = AppTrigger( - tenant_id=tenant_id, - app_id=app_model.id, - node_id="node-1", - trigger_type=AppTriggerType.TRIGGER_PLUGIN, - title="Trigger", - provider_name="provider", - status=status, - ) - session.add_all([app_model, trigger]) - session.commit() - return app_model, trigger +from models.model import App, AppMode +from models.trigger import AppTrigger, WorkflowWebhookTrigger +from tests.unit_tests.config_override import config_overrides_context +_CREATED_AT = datetime(2026, 1, 2, 3, 4, 5) -def test_parser_models_validate(): - parser = workflow_trigger_module.WebhookTriggerQuery(node_id="node-1") - enable_parser = workflow_trigger_module.ParserEnable( - trigger_id="550e8400-e29b-41d4-a716-446655440000", enable_trigger=True - ) - assert parser.node_id == "node-1" - assert enable_parser.enable_trigger is True - - -def test_workflow_trigger_response_serializes_datetime(): - created_at = datetime(2026, 1, 2, 3, 4, 5, tzinfo=UTC) - response = workflow_trigger_module.WorkflowTriggerResponse( - id=str(uuid.uuid4()), - trigger_type=AppTriggerType.TRIGGER_PLUGIN, - title="Trigger", - node_id="node-1", - provider_name="provider", - icon="https://example.com/icon", - status=AppTriggerStatus.ENABLED, - created_at=created_at, - updated_at=created_at, - ) +@dataclass(frozen=True) +class _Endpoint: + method: str + path: str + resource: type[Resource] - payload = response.model_dump(mode="json") - assert payload["id"] == response.id - assert payload["created_at"] == "2026-01-02T03:04:05Z" - assert payload["updated_at"] == "2026-01-02T03:04:05Z" +_ENDPOINTS = ( + _Endpoint("GET", "workflows/triggers/webhook", workflow_trigger.WebhookTriggerApi), + _Endpoint("GET", "triggers", workflow_trigger.AppTriggersApi), + _Endpoint("POST", "trigger-enable", workflow_trigger.AppTriggerEnableApi), +) -def test_webhook_trigger_response_serializes_datetime(): - created_at = datetime(2026, 1, 2, 3, 4, 5, tzinfo=UTC) - webhook = { - "id": "webhook-1", - "webhook_id": "whk-1", - "webhook_url": "https://example.com/hook", - "webhook_debug_url": "https://example.com/hook/debug", - "node_id": "node-1", - "created_at": created_at, - } - payload = workflow_trigger_module.WebhookTriggerResponse.model_validate(webhook).model_dump(mode="json") - assert payload["webhook_id"] == "whk-1" - assert payload["created_at"] == "2026-01-02T03:04:05Z" - - -def test_app_triggers_get_uses_injected_tenant_id(app: Flask, sqlite_session: Session) -> None: - app_model, trigger = _persist_app_trigger(sqlite_session) - other_tenant_trigger = AppTrigger( - tenant_id=str(uuid.uuid4()), - app_id=app_model.id, - node_id="node-1", - trigger_type=AppTriggerType.TRIGGER_PLUGIN, - title="Other Tenant Trigger", - provider_name="provider", - status=AppTriggerStatus.ENABLED, - ) - sqlite_session.add(other_tenant_trigger) - sqlite_session.commit() +@dataclass(frozen=True) +class _Harness: + client: FlaskClient + session_factory: sessionmaker[Session] + account_id: str + tenant_id: str + headers: dict[str, str] + sessions: list[Session] - api = workflow_trigger_module.AppTriggersApi() - method = inspect.unwrap(api.get) + def create_app(self, *, tenant_id: str | None = None, mode: AppMode = AppMode.WORKFLOW) -> str: + with self.session_factory.begin() as session: + app = App( + tenant_id=tenant_id or self.tenant_id, + name="Trigger app", + mode=mode, + created_by=self.account_id, + maintainer=self.account_id, + enable_api=True, + enable_site=True, + ) + session.add(app) + session.flush() + return app.id - with app.test_request_context("/"): - response = method(api, sqlite_session, app_model.tenant_id, app_model) + def create_trigger( + self, + app_id: str, + *, + tenant_id: str | None = None, + trigger_type: AppTriggerType = AppTriggerType.TRIGGER_PLUGIN, + status: AppTriggerStatus = AppTriggerStatus.ENABLED, + created_at: datetime = _CREATED_AT, + ) -> str: + with self.session_factory.begin() as session: + trigger = AppTrigger( + tenant_id=tenant_id or self.tenant_id, + app_id=app_id, + node_id="node-1", + trigger_type=trigger_type, + title="Trigger", + provider_name="langgenius/github/github", + status=status, + ) + trigger.created_at = created_at + trigger.updated_at = created_at + session.add(trigger) + session.flush() + return trigger.id - assert [item["id"] for item in response["data"]] == [trigger.id] - assert response["data"][0]["icon"].endswith("/provider/icon") + def create_webhook(self, app_id: str, *, tenant_id: str | None = None, node_id: str = "node-1") -> str: + with self.session_factory.begin() as session: + webhook = WorkflowWebhookTrigger( + app_id=app_id, + tenant_id=tenant_id or self.tenant_id, + node_id=node_id, + webhook_id=str(uuid4())[:24], + created_by=self.account_id, + ) + webhook.created_at = _CREATED_AT + session.add(webhook) + session.flush() + return webhook.id + def set_role(self, role: TenantAccountRole) -> None: + with self.session_factory.begin() as session: + membership = session.scalar( + select(TenantAccountJoin).where(TenantAccountJoin.account_id == self.account_id) + ) + assert membership is not None + membership.role = role -def test_app_trigger_enable_uses_injected_tenant_id(app: Flask, sqlite_session: Session) -> None: - app_model, trigger = _persist_app_trigger(sqlite_session, status=AppTriggerStatus.DISABLED) - payload = {"trigger_id": trigger.id, "enable_trigger": True} + def call(self, endpoint: _Endpoint, app_id: str) -> TestResponse: + self.sessions.clear() + if endpoint.method == "POST": + return self.client.post( + f"/console/api/apps/{app_id}/{endpoint.path}", + headers=self.headers, + json={"trigger_id": str(uuid4()), "enable_trigger": True}, + ) + return self.client.get( + f"/console/api/apps/{app_id}/{endpoint.path}", headers=self.headers, query_string={"node_id": "node-1"} + ) - api = workflow_trigger_module.AppTriggerEnableApi() - method = inspect.unwrap(api.post) + def enable(self, app_id: str, payload: dict[str, object]) -> TestResponse: + self.sessions.clear() + return self.client.post(f"/console/api/apps/{app_id}/trigger-enable", headers=self.headers, json=payload) + +@pytest.fixture +def harness(sqlite_engine: Engine, sqlite_session_factory: sessionmaker[Session]) -> Iterator[_Harness]: with ( - app.test_request_context("/", json=payload), - patch.object(type(console_ns), "payload", new_callable=PropertyMock, return_value=payload), + Redis() as redis_client, + config_overrides_context( + DEPLOYMENT_EDITION=DeploymentEdition.COMMUNITY, + RBAC_ENABLED=False, + LOGIN_DISABLED=False, + INIT_PASSWORD="", + SECRET_KEY="trigger-http-test-secret-at-least-32-bytes", + CONSOLE_API_URL="https://console.example.test", + ), ): - response = method( - api, - workflow_trigger_module.ParserEnable(trigger_id=trigger.id, enable_trigger=True), - sqlite_session, - app_model.tenant_id, - app_model, + app = Flask(__name__) + app.config.update(TESTING=True, SQLALCHEMY_DATABASE_URI=str(sqlite_engine.url)) + db.init_app(app) + redis = RedisClientWrapper() + redis.initialize(redis_client) + services = build_application_services( + database_client=sqlite_session_factory, + deployment_edition=DeploymentEdition.COMMUNITY, + initialization_password="", + redis=redis, ) + app.extensions["application_services"] = services + bind_account_loader(app, services.accounts.identity.load_user) + login_manager = DifyLoginManager() + login_manager.request_loader(load_user_from_request) + login_manager.unauthorized_handler(unauthorized_handler) + login_manager.init_app(app) + blueprint = Blueprint("console", __name__, url_prefix="/console/api") + api = ExternalApi(blueprint) + for endpoint in _ENDPOINTS: + api.add_resource(endpoint.resource, f"/apps//{endpoint.path}") + app.register_blueprint(blueprint) + + now = naive_utc_now() + with sqlite_session_factory.begin() as session: + account = Account( + email="triggers@example.com", + name="Trigger editor", + status=AccountStatus.ACTIVE, + initialized_at=now, + ) + account.last_active_at = now + tenant = Tenant(name="Trigger workspace") + session.add_all([account, tenant, DifySetup(version="test")]) + session.flush() + session.add( + TenantAccountJoin( + tenant_id=tenant.id, + account_id=account.id, + role=TenantAccountRole.OWNER, + current=True, + ) + ) + account_id, tenant_id = account.id, tenant.id + + # The legacy setup cache has a separate scoped session. Warm and remove + # it before observing the transaction boundaries of each actual request. + with app.app_context(): + _is_setup_completed.reset_success() + assert _is_setup_completed() + db.session.remove() + + client = app.test_client() + csrf_token = generate_csrf_token(account_id) + client.set_cookie(_real_cookie_name("csrf_token"), csrf_token) + headers = { + "Authorization": f"Bearer {PassportService().issue({'user_id': account_id})}", + HEADER_NAME_CSRF_TOKEN: csrf_token, + } + sessions: list[Session] = [] + + def observe_session(session: Session, _transaction: SessionTransaction, _connection: Connection) -> None: + sessions.append(session) + + @app.after_request + def assert_request_sessions_closed(response: Response) -> Response: + assert all(not session.in_transaction() and not session.identity_map for session in sessions) + return response + + event.listen(Session, "after_begin", observe_session) + try: + yield _Harness(client, sqlite_session_factory, account_id, tenant_id, headers, sessions) + finally: + event.remove(Session, "after_begin", observe_session) + _is_setup_completed.reset_success() + with app.app_context(): + db.session.remove() + db.engine.dispose() + + +@pytest.mark.parametrize("endpoint", _ENDPOINTS, ids=lambda endpoint: endpoint.path) +@pytest.mark.parametrize("visibility", ["missing", "different-workspace", "non-normal", "workflow-only", "wrong-mode"]) +def test_triggers_reject_inaccessible_apps(harness: _Harness, endpoint: _Endpoint, visibility: str) -> None: + app_id = str(uuid4()) + if visibility != "missing": + app_id = harness.create_app(tenant_id=str(uuid4()) if visibility == "different-workspace" else None) + with harness.session_factory.begin() as session: + app = session.get(App, app_id) + assert app is not None + if visibility == "non-normal": + session.execute(text("UPDATE apps SET status = 'archived' WHERE id = :app_id"), {"app_id": app_id}) + elif visibility == "wrong-mode": + app.mode = AppMode.CHAT + elif visibility == "workflow-only": + app.mode = AppMode.AGENT + session.add( + Agent( + tenant_id=harness.tenant_id, + name="Hidden workflow agent", + scope=AgentScope.WORKFLOW_ONLY, + source=AgentSource.WORKFLOW, + app_id=str(uuid4()), + backing_app_id=app_id, + ) + ) + + response = harness.call(endpoint, app_id) + + assert response.status_code == HTTPStatus.NOT_FOUND + assert response.get_json()["code"] == "app_not_found" + assert response.content_type == "application/json" + assert harness.sessions + + +def test_list_empty_triggers(harness: _Harness) -> None: + response = harness.call(_ENDPOINTS[1], harness.create_app()) + + assert response.status_code == HTTPStatus.OK + assert response.get_json() == {"data": []} + + +def test_list_orders_and_serializes_triggers_without_leaking_other_owners(harness: _Harness) -> None: + app_id = harness.create_app() + older = harness.create_trigger(app_id, created_at=datetime(2025, 1, 1)) + same_time = [ + harness.create_trigger(app_id, trigger_type=trigger_type, status=status) + for trigger_type, status in zip(AppTriggerType, AppTriggerStatus, strict=True) + ] + expected_statuses = dict(zip(same_time, AppTriggerStatus, strict=True)) + expected_statuses[older] = AppTriggerStatus.ENABLED + harness.create_trigger(app_id, tenant_id=str(uuid4())) + harness.create_trigger(harness.create_app()) + + response = harness.call(_ENDPOINTS[1], app_id) + + assert response.status_code == HTTPStatus.OK + data = response.get_json()["data"] + assert [item["id"] for item in data] == sorted(same_time, reverse=True) + [older] + for item in data: + assert set(item) == { + "id", + "trigger_type", + "title", + "node_id", + "provider_name", + "icon", + "status", + "created_at", + "updated_at", + } + assert item["title"] == "Trigger" + assert item["node_id"] == "node-1" + assert item["provider_name"] == "langgenius/github/github" + assert item["status"] == expected_statuses[item["id"]].value + if item["trigger_type"] == AppTriggerType.TRIGGER_PLUGIN: + assert item["icon"] == ( + "https://console.example.test/console/api/workspaces/current/tool-provider/builtin/" + "langgenius/github/github/icon" + ) + else: + assert item["icon"] == "" + if item["id"] in same_time: + assert item["created_at"] == "2026-01-02T03:04:05" + assert item["updated_at"] == "2026-01-02T03:04:05" + + +@pytest.mark.parametrize("role", [TenantAccountRole.OWNER, TenantAccountRole.ADMIN, TenantAccountRole.EDITOR]) +def test_editor_roles_can_disable_and_enable_with_committed_status(harness: _Harness, role: TenantAccountRole) -> None: + harness.set_role(role) + app_id = harness.create_app() + trigger_id = harness.create_trigger(app_id) + + for enabled, expected in [(False, AppTriggerStatus.DISABLED), (True, AppTriggerStatus.ENABLED)]: + response = harness.enable(app_id, {"trigger_id": trigger_id, "enable_trigger": enabled}) + + assert response.status_code == HTTPStatus.OK + body = response.get_json() + assert body["id"] == trigger_id + assert body["status"] == expected.value + assert body["icon"].endswith("/langgenius/github/github/icon") + with harness.session_factory() as session: + persisted = session.get(AppTrigger, trigger_id) + assert persisted is not None + assert persisted.status == expected + + +@pytest.mark.parametrize("role", [TenantAccountRole.NORMAL, TenantAccountRole.DATASET_OPERATOR]) +def test_read_only_roles_cannot_update_trigger(harness: _Harness, role: TenantAccountRole) -> None: + harness.set_role(role) + app_id = harness.create_app() + trigger_id = harness.create_trigger(app_id) + + response = harness.enable(app_id, {"trigger_id": trigger_id, "enable_trigger": False}) + + assert response.status_code == HTTPStatus.FORBIDDEN + with harness.session_factory() as session: + persisted = session.get(AppTrigger, trigger_id) + assert persisted is not None + assert persisted.status == AppTriggerStatus.ENABLED + assert harness.call(_ENDPOINTS[1], app_id).status_code == HTTPStatus.OK + + +@pytest.mark.parametrize("missing_field", ["trigger_id", "enable_trigger"]) +def test_invalid_update_payload_does_not_write(harness: _Harness, missing_field: str) -> None: + app_id = harness.create_app() + trigger_id = harness.create_trigger(app_id) + payload: dict[str, object] = {"trigger_id": trigger_id, "enable_trigger": False} + del payload[missing_field] + + response = harness.enable(app_id, payload) + + assert response.status_code == HTTPStatus.UNPROCESSABLE_ENTITY + with harness.session_factory() as session: + persisted = session.get(AppTrigger, trigger_id) + assert persisted is not None + assert persisted.status == AppTriggerStatus.ENABLED + + +@pytest.mark.parametrize("target", ["missing", "other-app", "other-tenant"]) +def test_update_rejects_trigger_outside_requested_owner_chain(harness: _Harness, target: str) -> None: + app_id = harness.create_app() + trigger_id = str(uuid4()) + if target == "other-app": + trigger_id = harness.create_trigger(harness.create_app()) + elif target == "other-tenant": + trigger_id = harness.create_trigger(app_id, tenant_id=str(uuid4())) + + response = harness.enable(app_id, {"trigger_id": trigger_id, "enable_trigger": False}) + + assert response.status_code == HTTPStatus.NOT_FOUND + assert response.get_json()["code"] == "app_trigger_not_found" + if target != "missing": + with harness.session_factory() as session: + persisted = session.get(AppTrigger, trigger_id) + assert persisted is not None + assert persisted.status == AppTriggerStatus.ENABLED + + +def test_webhook_returns_detached_configuration_and_urls(harness: _Harness) -> None: + app_id = harness.create_app() + webhook_id = harness.create_webhook(app_id) + with harness.session_factory() as session: + webhook = session.get(WorkflowWebhookTrigger, webhook_id) + assert webhook is not None + external_id = webhook.webhook_id + + response = harness.call(_ENDPOINTS[0], app_id) + + assert response.status_code == HTTPStatus.OK + assert response.get_json() == { + "id": webhook_id, + "webhook_id": external_id, + "node_id": "node-1", + "webhook_url": f"{str(base_url).rstrip('/')}/triggers/webhook/{external_id}", + "webhook_debug_url": f"{str(base_url).rstrip('/')}/triggers/webhook-debug/{external_id}", + "created_at": "2026-01-02T03:04:05", + } + + +@pytest.mark.parametrize("target", ["missing", "other-app", "other-tenant", "other-node"]) +def test_webhook_rejects_references_outside_requested_owner_chain(harness: _Harness, target: str) -> None: + app_id = harness.create_app() + if target == "other-app": + harness.create_webhook(harness.create_app()) + elif target == "other-tenant": + harness.create_webhook(app_id, tenant_id=str(uuid4())) + elif target == "other-node": + harness.create_webhook(app_id, node_id="node-2") + + response = harness.call(_ENDPOINTS[0], app_id) + + assert response.status_code == HTTPStatus.NOT_FOUND + assert response.get_json()["code"] == "webhook_trigger_not_found" + + +def test_app_owner_with_rbac_enabled_uses_real_owner_admission(harness: _Harness) -> None: + harness.set_role(TenantAccountRole.NORMAL) + app_id = harness.create_app() + trigger_id = harness.create_trigger(app_id) + + with config_overrides_context(RBAC_ENABLED=True): + response = harness.enable(app_id, {"trigger_id": trigger_id, "enable_trigger": False}) + + assert response.status_code == HTTPStatus.OK + assert response.get_json()["status"] == AppTriggerStatus.DISABLED + + +def test_update_requires_csrf_before_persistence(harness: _Harness) -> None: + app_id = harness.create_app() + trigger_id = harness.create_trigger(app_id) + harness.sessions.clear() + + response = harness.client.post( + f"/console/api/apps/{app_id}/trigger-enable", + headers={"Authorization": harness.headers["Authorization"]}, + json={"trigger_id": trigger_id, "enable_trigger": False}, + ) - assert response["id"] == trigger.id - assert response["status"] == "enabled" - sqlite_session.flush() - sqlite_session.expire_all() - persisted_trigger = sqlite_session.get(AppTrigger, trigger.id) - assert persisted_trigger is not None - assert persisted_trigger.status == AppTriggerStatus.ENABLED + assert response.status_code == HTTPStatus.UNAUTHORIZED + with harness.session_factory() as session: + persisted = session.get(AppTrigger, trigger_id) + assert persisted is not None + assert persisted.status == AppTriggerStatus.ENABLED diff --git a/api/tests/unit_tests/controllers/trigger/test_webhook.py b/api/tests/unit_tests/controllers/trigger/test_webhook.py index 0e80e896b94452..760e94e3b1dbf5 100644 --- a/api/tests/unit_tests/controllers/trigger/test_webhook.py +++ b/api/tests/unit_tests/controllers/trigger/test_webhook.py @@ -2,12 +2,15 @@ from unittest.mock import patch import pytest +from sqlalchemy.orm import Session, sessionmaker from werkzeug.exceptions import NotFound, RequestEntityTooLarge import controllers.trigger.webhook as module from models.trigger import WorkflowWebhookTrigger from models.workflow import Workflow +from repositories.trigger.app_repository import SQLAlchemyAppTriggerRepository from services.errors.app import QuotaExceededError +from services.trigger.app_trigger_service import AppTriggerService from tests.unit_tests.model_factories import make_workflow @@ -22,9 +25,15 @@ class _AppScopedEndUserServicesStub: commands: object +@dataclass(frozen=True) +class _AppServicesStub: + trigger_status: AppTriggerService + + class _ApplicationServicesStub: - def __init__(self, commands: object) -> None: + def __init__(self, commands: object, trigger_status: AppTriggerService) -> None: self.app_scoped_end_users = _AppScopedEndUserServicesStub(commands) + self.apps = _AppServicesStub(trigger_status) @pytest.fixture(autouse=True) @@ -38,9 +47,10 @@ def mock_jsonify(): @pytest.fixture(autouse=True) -def end_user_commands(monkeypatch: pytest.MonkeyPatch) -> object: +def end_user_commands(monkeypatch: pytest.MonkeyPatch, sqlite_session_factory: sessionmaker[Session]) -> object: commands = object() - monkeypatch.setattr(module, "application_services", lambda: _ApplicationServicesStub(commands)) + triggers = AppTriggerService(triggers=SQLAlchemyAppTriggerRepository(session_factory=sqlite_session_factory)) + monkeypatch.setattr(module, "application_services", lambda: _ApplicationServicesStub(commands, triggers)) return commands diff --git a/api/tests/unit_tests/events/event_handlers/test_app_trigger_publication.py b/api/tests/unit_tests/events/event_handlers/test_app_trigger_publication.py new file mode 100644 index 00000000000000..8ca39693d93437 --- /dev/null +++ b/api/tests/unit_tests/events/event_handlers/test_app_trigger_publication.py @@ -0,0 +1,167 @@ +"""Application trigger registration shares publication's real SQL transaction.""" + +from uuid import uuid4 + +import pytest +from sqlalchemy import select +from sqlalchemy.orm import Session, sessionmaker + +from events.event_handlers.update_app_triggers_when_app_published_workflow_updated import handle +from models.enums import AppTriggerStatus, AppTriggerType +from models.model import App, AppMode +from models.trigger import AppTrigger +from models.workflow import Workflow +from tests.unit_tests.events.event_handlers.test_sync_plugin_trigger_when_app_created import ( + _app, + _workflow, + binding_app, +) + +__all__ = ["binding_app"] + + +def _node( + node_id: str, + *, + title: str = "Published trigger", + kind: str = AppTriggerType.TRIGGER_PLUGIN, + provider: str = "github", +) -> dict[str, object]: + return {"id": node_id, "data": {"type": kind, "title": title, "provider_name": provider}} + + +def _records(session: Session, app: App) -> list[AppTrigger]: + return list( + session.scalars( + select(AppTrigger) + .where(AppTrigger.tenant_id == app.tenant_id, AppTrigger.app_id == app.id) + .order_by(AppTrigger.node_id) + ) + ) + + +@pytest.mark.parametrize("commit", [True, False], ids=["commit", "rollback"]) +def test_publication_and_registration_share_callers_transaction( + sqlite_session_factory: sessionmaker[Session], commit: bool +) -> None: + app = _app() + with sqlite_session_factory.begin() as session: + session.add(app) + workflow = _workflow(app, [_node("trigger")]) + with sqlite_session_factory() as session: + session.add(workflow) + stored = session.get_one(App, app.id) + stored.workflow_id = workflow.id + handle(stored, session=session, published_workflow=workflow) + registered = _records(session, stored) + assert len(registered) == 1 + assert registered[0].status == AppTriggerStatus.ENABLED + assert session.in_transaction() + if commit: + session.commit() + else: + session.rollback() + with sqlite_session_factory() as session: + assert (session.get(Workflow, workflow.id) is not None) == commit + assert session.get_one(App, app.id).workflow_id == (workflow.id if commit else None) + assert len(_records(session, app)) == int(commit) + + +def test_sync_preserves_existing_status_and_metadata_and_updates_nonempty_titles( + sqlite_session_factory: sessionmaker[Session], +) -> None: + app = _app() + with sqlite_session_factory.begin() as session: + session.add(app) + records = [ + AppTrigger( + tenant_id=app.tenant_id, + app_id=app.id, + node_id=node_id, + trigger_type=AppTriggerType.TRIGGER_WEBHOOK, + title=f"Existing {node_id}", + provider_name="original", + status=status, + ) + for node_id, status in [ + ("renamed", AppTriggerStatus.DISABLED), + ("blank", AppTriggerStatus.RATE_LIMITED), + ("removed", AppTriggerStatus.ENABLED), + ] + ] + session.add_all(records) + session.flush() + ids = {record.node_id: record.id for record in records} + workflow = _workflow(app, [_node("renamed", title="New name"), _node("blank", title=""), _node("added")]) + with sqlite_session_factory.begin() as session: + handle(app, session=session, published_workflow=workflow) + with sqlite_session_factory() as session: + stored = {record.node_id: record for record in _records(session, app)} + assert set(stored) == {"renamed", "blank", "added"} + assert stored["renamed"].id == ids["renamed"] + assert stored["renamed"].title == "New name" + assert stored["renamed"].status == AppTriggerStatus.DISABLED + assert stored["renamed"].trigger_type == AppTriggerType.TRIGGER_WEBHOOK + assert stored["renamed"].provider_name == "original" + assert stored["blank"].title == "Existing blank" + assert stored["blank"].status == AppTriggerStatus.RATE_LIMITED + assert stored["added"].status == AppTriggerStatus.ENABLED + assert stored["added"].trigger_type == AppTriggerType.TRIGGER_PLUGIN + assert stored["added"].provider_name == "github" + + +@pytest.mark.parametrize("foreign_field", ["tenant_id", "app_id"]) +def test_sync_does_not_delete_or_modify_foreign_trigger( + sqlite_session_factory: sessionmaker[Session], foreign_field: str +) -> None: + app = _app() + foreign = AppTrigger( + tenant_id=str(uuid4()) if foreign_field == "tenant_id" else app.tenant_id, + app_id=str(uuid4()) if foreign_field == "app_id" else app.id, + node_id="foreign", + trigger_type=AppTriggerType.TRIGGER_PLUGIN, + title="Foreign trigger", + status=AppTriggerStatus.DISABLED, + ) + with sqlite_session_factory.begin() as session: + session.add_all([app, foreign]) + with sqlite_session_factory.begin() as session: + handle(app, session=session, published_workflow=_workflow(app, [])) + with sqlite_session_factory() as session: + stored = session.get_one(AppTrigger, foreign.id) + assert stored.title == "Foreign trigger" + assert stored.status == AppTriggerStatus.DISABLED + + +def test_only_trigger_nodes_are_registered(sqlite_session_factory: sessionmaker[Session]) -> None: + app = _app() + nodes = [_node("start", kind="start"), _node("llm", kind="llm")] + nodes.extend( + _node(str(index), kind=kind) + for index, kind in enumerate( + [AppTriggerType.TRIGGER_PLUGIN, AppTriggerType.TRIGGER_WEBHOOK, AppTriggerType.TRIGGER_SCHEDULE] + ) + ) + with sqlite_session_factory.begin() as session: + session.add(app) + handle(app, session=session, published_workflow=_workflow(app, nodes)) + with sqlite_session_factory() as session: + records = _records(session, app) + assert {record.node_id for record in records} == {"0", "1", "2"} + assert {record.status for record in records} == {AppTriggerStatus.ENABLED} + + +@pytest.mark.parametrize("mode", [AppMode.CHAT, AppMode.COMPLETION]) +def test_non_workflow_app_is_ignored(sqlite_session_factory: sessionmaker[Session], mode: AppMode) -> None: + app = _app() + app.mode = mode + with sqlite_session_factory.begin() as session: + handle(app, session=session, published_workflow=_workflow(app, [_node("trigger")])) + assert _records(session, app) == [] + + +def test_missing_published_workflow_is_ignored(sqlite_session_factory: sessionmaker[Session]) -> None: + app = _app() + with sqlite_session_factory.begin() as session: + handle(app, session=session) + assert _records(session, app) == [] diff --git a/api/tests/unit_tests/pyrefly.toml b/api/tests/unit_tests/pyrefly.toml index 8b9e33c6c6eaec..7b0348a7edc642 100644 --- a/api/tests/unit_tests/pyrefly.toml +++ b/api/tests/unit_tests/pyrefly.toml @@ -42,7 +42,6 @@ project-excludes = [ "controllers/console/app/test_workflow_node_output_inspector.py", "controllers/console/app/test_workflow_pause_details_api.py", "controllers/console/app/test_workflow_run_api.py", - "controllers/console/app/test_workflow_trigger_api.py", "controllers/console/app/test_wraps.py", "controllers/console/app/workflow_draft_variables_test.py", "controllers/console/auth/test_account_activation.py", diff --git a/api/tests/unit_tests/repositories/test_app_trigger_repository.py b/api/tests/unit_tests/repositories/test_app_trigger_repository.py new file mode 100644 index 00000000000000..95cb67da928760 --- /dev/null +++ b/api/tests/unit_tests/repositories/test_app_trigger_repository.py @@ -0,0 +1,389 @@ +"""Exercise app trigger ownership and transaction boundaries with real SQLite.""" + +from collections.abc import Iterator +from dataclasses import dataclass +from datetime import datetime, timedelta +from functools import partial +from uuid import uuid4 + +import pytest +from sqlalchemy import Engine, event, select, text +from sqlalchemy.exc import IntegrityError +from sqlalchemy.orm import Session, sessionmaker + +from machinery.context import RequestContext +from models.agent import Agent, AgentScope, AgentSource, AgentStatus +from models.enums import AppTriggerStatus, AppTriggerType +from models.model import App, AppMode +from models.trigger import AppTrigger, WorkflowWebhookTrigger +from repositories.trigger.app_repository import SQLAlchemyAppTriggerRepository +from services.app.console_service import ConsoleAppNotFoundError +from services.trigger.app_trigger_store import AppTriggerNotFoundError, WebhookTriggerNotFoundError + +_CREATED_AT = datetime(2026, 10, 20, 8, 0) + + +@dataclass(frozen=True) +class TriggerScope: + context: RequestContext + app_id: str + trigger_id: str + webhook_id: str + node_id: str + + +@pytest.fixture +def scope(sqlite_session_factory: sessionmaker[Session]) -> TriggerScope: + context = RequestContext("request", None, str(uuid4()), str(uuid4())) + with sqlite_session_factory.begin() as session: + app = App( + tenant_id=context.active_workspace_id, + name="Workflow", + mode=AppMode.WORKFLOW, + enable_site=True, + enable_api=True, + ) + session.add(app) + session.flush() + trigger = AppTrigger( + tenant_id=context.active_workspace_id, + app_id=app.id, + node_id="webhook-node", + trigger_type=AppTriggerType.TRIGGER_WEBHOOK, + title="Webhook", + provider_name=None, + status=AppTriggerStatus.ENABLED, + ) + webhook = WorkflowWebhookTrigger( + tenant_id=context.active_workspace_id, + app_id=app.id, + node_id=trigger.node_id or "", + webhook_id="webhook-endpoint", + created_by=context.account_id, + ) + trigger.created_at = _CREATED_AT + webhook.created_at = _CREATED_AT + session.add_all([trigger, webhook]) + session.flush() + # The insert default replaces None with ""; existing rows may still be NULL. + trigger.provider_name = None + session.flush() + return TriggerScope(context, app.id, trigger.id, webhook.id, webhook.node_id) + + +@pytest.fixture +def repository(sqlite_session_factory: sessionmaker[Session]) -> SQLAlchemyAppTriggerRepository: + return SQLAlchemyAppTriggerRepository(session_factory=sqlite_session_factory) + + +@pytest.fixture +def open_connections(sqlite_engine: Engine) -> Iterator[set[object]]: + connections: set[object] = set() + + def checkout(connection: object, *_args: object) -> None: + connections.add(connection) + + def checkin(connection: object, *_args: object) -> None: + connections.remove(connection) + + event.listen(sqlite_engine, "checkout", checkout) + event.listen(sqlite_engine, "checkin", checkin) + try: + yield connections + assert not connections + finally: + event.remove(sqlite_engine, "checkout", checkout) + event.remove(sqlite_engine, "checkin", checkin) + + +def test_reads_return_detached_records_and_release_connections( + repository: SQLAlchemyAppTriggerRepository, + scope: TriggerScope, + open_connections: set[object], +) -> None: + triggers = repository.list_triggers(scope.context, scope.app_id) + assert not open_connections + webhook = repository.get_webhook(scope.context, scope.app_id, scope.node_id) + assert not open_connections + + assert len(triggers) == 1 + trigger = triggers[0] + assert trigger.id == scope.trigger_id + assert trigger.trigger_type == AppTriggerType.TRIGGER_WEBHOOK + assert trigger.title == "Webhook" + assert trigger.node_id == scope.node_id + assert trigger.provider_name is None + assert trigger.status == AppTriggerStatus.ENABLED + assert trigger.created_at == _CREATED_AT + assert webhook.id == scope.webhook_id + assert webhook.webhook_id == "webhook-endpoint" + assert webhook.node_id == scope.node_id + assert webhook.created_at == _CREATED_AT + + repository.set_enabled(scope.context, scope.app_id, scope.trigger_id, enabled=False) + # Existing query results remain snapshots even after the database changes. + assert trigger.status == AppTriggerStatus.ENABLED + assert not open_connections + + +@pytest.mark.parametrize("operation", ["get_webhook", "list_triggers", "set_enabled"]) +@pytest.mark.parametrize("visibility", ["missing", "other_workspace", "disabled", "wrong_mode", "hidden_agent"]) +def test_operations_require_visible_workflow_in_current_workspace( + repository: SQLAlchemyAppTriggerRepository, + sqlite_session_factory: sessionmaker[Session], + scope: TriggerScope, + operation: str, + visibility: str, + open_connections: set[object], +) -> None: + context = scope.context + app_id = scope.app_id + with sqlite_session_factory.begin() as session: + if visibility == "missing": + app_id = str(uuid4()) + elif visibility == "other_workspace": + context = context._replace(active_workspace_id=str(uuid4())) + elif visibility == "disabled": + session.execute(text("UPDATE apps SET status = 'disabled' WHERE id = :id"), {"id": app_id}) + elif visibility == "wrong_mode": + session.get_one(App, app_id).mode = AppMode.CHAT + else: + session.get_one(App, app_id).mode = AppMode.AGENT + session.add( + Agent( + tenant_id=context.active_workspace_id, + name="Workflow-only", + scope=AgentScope.WORKFLOW_ONLY, + source=AgentSource.WORKFLOW, + status=AgentStatus.ACTIVE, + backing_app_id=app_id, + ) + ) + + if operation == "get_webhook": + run = partial(repository.get_webhook, context, app_id, scope.node_id) + elif operation == "list_triggers": + run = partial(repository.list_triggers, context, app_id) + else: + run = partial(repository.set_enabled, context, app_id, scope.trigger_id, enabled=False) + with pytest.raises(ConsoleAppNotFoundError) as raised: + run() + assert not open_connections + if visibility == "wrong_mode": + assert str(raised.value) == "App mode is not in the supported list: {'workflow'}" + else: + assert str(raised.value) == "App not found" + with sqlite_session_factory() as session: + assert session.get_one(AppTrigger, scope.trigger_id).status == AppTriggerStatus.ENABLED + + +@pytest.mark.parametrize("field", ["tenant_id", "app_id", "node_id"]) +def test_webhook_lookup_requires_complete_owner_chain_and_node( + repository: SQLAlchemyAppTriggerRepository, + sqlite_session_factory: sessionmaker[Session], + scope: TriggerScope, + field: str, +) -> None: + with sqlite_session_factory.begin() as session: + webhook = session.get_one(WorkflowWebhookTrigger, scope.webhook_id) + if field == "tenant_id": + webhook.tenant_id = str(uuid4()) + elif field == "app_id": + webhook.app_id = str(uuid4()) + else: + webhook.node_id = "other-node" + + with pytest.raises(WebhookTriggerNotFoundError, match="Webhook trigger not found for this node"): + repository.get_webhook(scope.context, scope.app_id, scope.node_id) + + +def test_list_preserves_stable_order_and_excludes_foreign_records( + repository: SQLAlchemyAppTriggerRepository, + sqlite_session_factory: sessionmaker[Session], + scope: TriggerScope, +) -> None: + with sqlite_session_factory.begin() as session: + older = session.get_one(AppTrigger, scope.trigger_id) + records = [ + AppTrigger( + tenant_id=scope.context.active_workspace_id, + app_id=scope.app_id, + node_id=f"node-{index}", + trigger_type=trigger_type, + title=f"Trigger {index}", + provider_name="provider" if trigger_type == AppTriggerType.TRIGGER_PLUGIN else "", + ) + for index, trigger_type in enumerate( + [AppTriggerType.TRIGGER_PLUGIN, AppTriggerType.TRIGGER_SCHEDULE, AppTriggerType.UNKNOWN] + ) + ] + for record in records: + record.created_at = _CREATED_AT + timedelta(days=1) + foreign_workspace = AppTrigger( + tenant_id=str(uuid4()), + app_id=scope.app_id, + node_id="foreign-workspace", + trigger_type=AppTriggerType.TRIGGER_WEBHOOK, + title="Foreign workspace", + ) + foreign_app = AppTrigger( + tenant_id=scope.context.active_workspace_id, + app_id=str(uuid4()), + node_id="foreign-app", + trigger_type=AppTriggerType.TRIGGER_WEBHOOK, + title="Foreign app", + ) + session.add_all([*records, foreign_workspace, foreign_app]) + session.flush() + expected_ids = sorted([record.id for record in records], reverse=True) + [older.id] + + result = repository.list_triggers(scope.context, scope.app_id) + + assert [record.id for record in result] == expected_ids + assert {record.trigger_type for record in result} == set(AppTriggerType) + + +def test_empty_app_has_no_triggers( + repository: SQLAlchemyAppTriggerRepository, sqlite_session_factory: sessionmaker[Session], scope: TriggerScope +) -> None: + with sqlite_session_factory.begin() as session: + session.delete(session.get_one(AppTrigger, scope.trigger_id)) + + assert repository.list_triggers(scope.context, scope.app_id) == () + + +@pytest.mark.parametrize("enabled", [True, False]) +@pytest.mark.parametrize("initial_status", list(AppTriggerStatus)) +def test_enable_disable_commits_only_status_and_returns_persisted_timestamps( + repository: SQLAlchemyAppTriggerRepository, + sqlite_session_factory: sessionmaker[Session], + scope: TriggerScope, + enabled: bool, + initial_status: AppTriggerStatus, + open_connections: set[object], +) -> None: + with sqlite_session_factory.begin() as session: + session.get_one(AppTrigger, scope.trigger_id).status = initial_status + + result = repository.set_enabled(scope.context, scope.app_id, scope.trigger_id, enabled=enabled) + assert not open_connections + + with sqlite_session_factory() as session: + stored = session.get_one(AppTrigger, scope.trigger_id) + assert stored.status == (AppTriggerStatus.ENABLED if enabled else AppTriggerStatus.DISABLED) + assert result.status == stored.status + assert result.created_at == stored.created_at == _CREATED_AT + assert result.updated_at == stored.updated_at + assert stored.tenant_id == scope.context.active_workspace_id + assert stored.app_id == scope.app_id + assert stored.title == "Webhook" + assert stored.node_id == scope.node_id + assert stored.provider_name is None + assert session.scalar(select(WorkflowWebhookTrigger.id)) == scope.webhook_id + + +@pytest.mark.parametrize("field", ["tenant_id", "app_id", "id"]) +def test_status_update_requires_complete_owner_chain_and_trigger( + repository: SQLAlchemyAppTriggerRepository, + sqlite_session_factory: sessionmaker[Session], + scope: TriggerScope, + field: str, +) -> None: + trigger_id = scope.trigger_id + with sqlite_session_factory.begin() as session: + trigger = session.get_one(AppTrigger, trigger_id) + if field == "tenant_id": + trigger.tenant_id = str(uuid4()) + elif field == "app_id": + trigger.app_id = str(uuid4()) + else: + trigger_id = str(uuid4()) + + with pytest.raises(AppTriggerNotFoundError, match="Trigger not found"): + repository.set_enabled(scope.context, scope.app_id, trigger_id, enabled=False) + + with sqlite_session_factory() as session: + assert session.get_one(AppTrigger, scope.trigger_id).status == AppTriggerStatus.ENABLED + + +def test_failed_status_write_rolls_back_and_releases_connection( + repository: SQLAlchemyAppTriggerRepository, + sqlite_session_factory: sessionmaker[Session], + scope: TriggerScope, + open_connections: set[object], +) -> None: + with sqlite_session_factory.begin() as session: + session.execute( + text( + "CREATE TRIGGER reject_trigger_update BEFORE UPDATE ON app_triggers " + "BEGIN SELECT RAISE(ABORT, 'trigger update rejected'); END" + ) + ) + + with pytest.raises(IntegrityError, match="trigger update rejected"): + repository.set_enabled(scope.context, scope.app_id, scope.trigger_id, enabled=False) + assert not open_connections + with sqlite_session_factory() as session: + assert session.get_one(AppTrigger, scope.trigger_id).status == AppTriggerStatus.ENABLED + + +@pytest.mark.parametrize("initial_status", list(AppTriggerStatus)) +def test_quota_status_updates_only_enabled_triggers_in_tenant( + repository: SQLAlchemyAppTriggerRepository, + sqlite_session_factory: sessionmaker[Session], + scope: TriggerScope, + initial_status: AppTriggerStatus, + open_connections: set[object], +) -> None: + with sqlite_session_factory.begin() as session: + session.get_one(AppTrigger, scope.trigger_id).status = initial_status + other_app = AppTrigger( + tenant_id=scope.context.active_workspace_id, + app_id=str(uuid4()), + node_id="other-app", + trigger_type=AppTriggerType.TRIGGER_SCHEDULE, + title="Same tenant", + status=AppTriggerStatus.ENABLED, + ) + foreign = AppTrigger( + tenant_id=str(uuid4()), + app_id=scope.app_id, + node_id=scope.node_id, + trigger_type=AppTriggerType.TRIGGER_PLUGIN, + title="Other tenant", + status=AppTriggerStatus.ENABLED, + ) + session.add_all([other_app, foreign]) + repository.mark_tenant_triggers_rate_limited(scope.context.active_workspace_id) + assert not open_connections + with sqlite_session_factory() as session: + expected = AppTriggerStatus.RATE_LIMITED if initial_status == AppTriggerStatus.ENABLED else initial_status + assert session.get_one(AppTrigger, scope.trigger_id).status == expected + assert session.get_one(AppTrigger, other_app.id).status == AppTriggerStatus.RATE_LIMITED + assert session.get_one(AppTrigger, foreign.id).status == AppTriggerStatus.ENABLED + + +def test_best_effort_status_failure_rolls_back_and_preserves_original_error_path( + repository: SQLAlchemyAppTriggerRepository, + sqlite_session_factory: sessionmaker[Session], + scope: TriggerScope, + open_connections: set[object], + caplog: pytest.LogCaptureFixture, +) -> None: + from services.trigger.app_trigger_service import AppTriggerService + + with sqlite_session_factory.begin() as session: + session.execute( + text( + "CREATE TRIGGER reject_trigger_update BEFORE UPDATE ON app_triggers " + "BEGIN SELECT RAISE(ABORT, 'status unavailable'); END" + ) + ) + AppTriggerService(triggers=repository).mark_tenant_triggers_rate_limited(scope.context.active_workspace_id) + assert not open_connections + assert ( + f"Failed to mark all enabled triggers as rate limited for tenant {scope.context.active_workspace_id}" + in caplog.text + ) + with sqlite_session_factory() as session: + assert session.get_one(AppTrigger, scope.trigger_id).status == AppTriggerStatus.ENABLED diff --git a/api/tests/unit_tests/services/test_webhook_service.py b/api/tests/unit_tests/services/test_webhook_service.py index 09954f977b0525..ac7392fa303bb8 100644 --- a/api/tests/unit_tests/services/test_webhook_service.py +++ b/api/tests/unit_tests/services/test_webhook_service.py @@ -6,7 +6,7 @@ import pytest from flask import Flask from sqlalchemy.engine import Engine -from sqlalchemy.orm import Session +from sqlalchemy.orm import Session, sessionmaker from werkzeug.datastructures import FileStorage from models.enums import AppTriggerStatus, AppTriggerType, EndUserType @@ -14,12 +14,19 @@ from models.tools import ToolFile from models.trigger import AppTrigger, WorkflowWebhookTrigger from models.workflow import Workflow +from repositories.trigger.app_repository import SQLAlchemyAppTriggerRepository from services.errors.app import QuotaExceededError from services.trigger import webhook_service as webhook_service_module +from services.trigger.app_trigger_service import AppTriggerService from services.trigger.webhook_service import WebhookService from tests.unit_tests.model_factories import make_app, make_end_user, make_workflow +@pytest.fixture +def trigger_status(sqlite_session_factory: sessionmaker[Session]) -> AppTriggerService: + return AppTriggerService(triggers=SQLAlchemyAppTriggerRepository(session_factory=sqlite_session_factory)) + + class _EndUserServiceStub: def __init__(self, result: EndUser | Exception) -> None: self._result = result @@ -234,6 +241,7 @@ def test_trigger_workflow_execution_propagates_quota_error_without_error_log( caplog: pytest.LogCaptureFixture, monkeypatch: pytest.MonkeyPatch, sqlite_engine: Engine, + trigger_status: AppTriggerService, ) -> None: """Quota failures refund the charge and close the real service-owned session.""" monkeypatch.setattr(webhook_service_module, "db", SimpleNamespace(engine=sqlite_engine)) @@ -256,6 +264,7 @@ def test_trigger_workflow_execution_propagates_quota_error_without_error_log( {"body": {}, "headers": {}, "query_params": {}, "files": {}, "method": "POST"}, workflow, end_users=_EndUserServiceStub(_end_user()), + trigger_status=trigger_status, ) assert exc_info.value is quota_error @@ -273,6 +282,7 @@ def test_trigger_workflow_execution_success( self, monkeypatch: pytest.MonkeyPatch, sqlite_engine: Engine, + trigger_status: AppTriggerService, ) -> None: monkeypatch.setattr(webhook_service_module, "db", SimpleNamespace(engine=sqlite_engine)) webhook_trigger = _webhook_trigger() @@ -299,6 +309,7 @@ def test_trigger_workflow_execution_success( webhook_data, workflow, end_users=_EndUserServiceStub(end_user), + trigger_status=trigger_status, ) call_session = mock_trigger.call_args.kwargs["session"] @@ -306,7 +317,7 @@ def test_trigger_workflow_execution_success( quota_charge.commit.assert_called_once_with() quota_charge.refund.assert_not_called() - def test_trigger_workflow_execution_end_user_service_failure(self) -> None: + def test_trigger_workflow_execution_end_user_service_failure(self, trigger_status: AppTriggerService) -> None: webhook_trigger = _webhook_trigger() workflow = _workflow() webhook_data = {"method": "POST", "headers": {}, "query_params": {}, "body": {}, "files": {}} @@ -317,6 +328,7 @@ def test_trigger_workflow_execution_end_user_service_failure(self) -> None: webhook_data, workflow, end_users=_EndUserServiceStub(ValueError("Failed to create end user")), + trigger_status=trigger_status, ) def test_extract_webhook_data_json(self): diff --git a/api/tests/unit_tests/tasks/test_trigger_processing_tasks.py b/api/tests/unit_tests/tasks/test_trigger_processing_tasks.py index 2403108b9f1b7a..582ca337f3e66f 100644 --- a/api/tests/unit_tests/tasks/test_trigger_processing_tasks.py +++ b/api/tests/unit_tests/tasks/test_trigger_processing_tasks.py @@ -134,6 +134,7 @@ def invoke() -> int: end_users=services.app_scoped_end_users.commands, events=services.workspaces.trigger_events, bindings=services.workspaces.trigger_binding_queries, + trigger_status=services.apps.trigger_status, ) return Context().run(invoke) diff --git a/packages/contracts/generated/api/console/apps/orpc.gen.ts b/packages/contracts/generated/api/console/apps/orpc.gen.ts index b19f277e3c2114..1bcb24be6b25e4 100644 --- a/packages/contracts/generated/api/console/apps/orpc.gen.ts +++ b/packages/contracts/generated/api/console/apps/orpc.gen.ts @@ -13,6 +13,7 @@ import { zAgentVersionNotFoundError, zAppNotFoundError, zAppSuggestedQuestionsAfterAnswerDisabledError, + zAppTriggerNotFoundError, zAppUnavailableError, zAudioTooLargeError, zBindingFileDownloadUnavailableError, @@ -51,6 +52,7 @@ import { zTracingProviderUnavailableError, zUnsupportedAudioTypeError, zUnsupportedTracingProviderError, + zWebhookTriggerNotFoundError, zWorkflowRunNotFoundError, } from './zod.gen.ts' import { @@ -2834,7 +2836,10 @@ export const post37 = oc }), ) .output(zPostAppsByAppIdTriggerEnableResponse) - .errors({ app_not_found: { status: 404, data: zAppNotFoundError } }) + .errors({ + app_not_found: { status: 404, data: zAppNotFoundError }, + app_trigger_not_found: { status: 404, data: zAppTriggerNotFoundError }, + }) export const triggerEnable = { post: post37, @@ -4820,7 +4825,10 @@ export const get84 = oc }), ) .output(zGetAppsByAppIdWorkflowsTriggersWebhookResponse) - .errors({ app_not_found: { status: 404, data: zAppNotFoundError } }) + .errors({ + app_not_found: { status: 404, data: zAppNotFoundError }, + webhook_trigger_not_found: { status: 404, data: zWebhookTriggerNotFoundError }, + }) export const webhook = { get: get84, diff --git a/packages/contracts/generated/api/console/apps/types.gen.ts b/packages/contracts/generated/api/console/apps/types.gen.ts index 96355856cfe014..baad20529c04ff 100644 --- a/packages/contracts/generated/api/console/apps/types.gen.ts +++ b/packages/contracts/generated/api/console/apps/types.gen.ts @@ -934,6 +934,11 @@ export type WorkflowTriggerResponse = { updated_at?: string | null } +export type AppTriggerNotFoundError = ErrorResponse & { + code: 'app_trigger_not_found' + status: 404 +} + export type WorkflowTriggerListResponse = { data: Array } @@ -1446,6 +1451,11 @@ export type WebhookTriggerResponse = { webhook_url: string } +export type WebhookTriggerNotFoundError = ErrorResponse & { + code: 'webhook_trigger_not_found' + status: 404 +} + export type WorkflowUpdatePayload = { marked_comment?: string | null marked_name?: string | null @@ -7182,7 +7192,7 @@ export type PostAppsByAppIdTriggerEnableData = { } export type PostAppsByAppIdTriggerEnableErrors = { - 404: AppNotFoundError | ErrorResponse + 404: AppNotFoundError | AppTriggerNotFoundError | ErrorResponse default: ErrorResponse } @@ -9121,7 +9131,7 @@ export type GetAppsByAppIdWorkflowsTriggersWebhookData = { } export type GetAppsByAppIdWorkflowsTriggersWebhookErrors = { - 404: AppNotFoundError | ErrorResponse + 404: AppNotFoundError | WebhookTriggerNotFoundError | ErrorResponse default: ErrorResponse } diff --git a/packages/contracts/generated/api/console/apps/zod.gen.ts b/packages/contracts/generated/api/console/apps/zod.gen.ts index e6cc027d761394..c40507b86d6fd0 100644 --- a/packages/contracts/generated/api/console/apps/zod.gen.ts +++ b/packages/contracts/generated/api/console/apps/zod.gen.ts @@ -883,6 +883,18 @@ export const zWorkflowTriggerResponse = z.object({ updated_at: z.iso.datetime().nullish(), }) +/** + * AppTriggerNotFoundError + * + * Trigger not found. + */ +export const zAppTriggerNotFoundError = zErrorResponse.and( + z.object({ + code: z.literal('app_trigger_not_found'), + status: z.literal(404), + }), +) + /** * WorkflowTriggerListResponse */ @@ -1273,6 +1285,18 @@ export const zWebhookTriggerResponse = z.object({ webhook_url: z.string(), }) +/** + * WebhookTriggerNotFoundError + * + * Webhook trigger not found for this node. + */ +export const zWebhookTriggerNotFoundError = zErrorResponse.and( + z.object({ + code: z.literal('webhook_trigger_not_found'), + status: z.literal(404), + }), +) + /** * WorkflowUpdatePayload */