diff --git a/api/.importlinter b/api/.importlinter index d7537957bab..6c9c709ea73 100644 --- a/api/.importlinter +++ b/api/.importlinter @@ -1275,6 +1275,7 @@ 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 diff --git a/api/controllers/trigger/webhook.py b/api/controllers/trigger/webhook.py index b9734cb9f39..a2f46418de2 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 d55fe262fb2..566c1fba71f 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 14f10db69a4..9491683b3a5 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 @@ -33,6 +35,7 @@ from services.app_tracing_config_service import AppTracingConfigService 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 @@ -46,6 +49,8 @@ class AppServices: console: ConsoleAppService enterprise: EnterpriseAppService triggers: AppTriggerStore + trigger_status: AppTriggerService + create_trigger_publication_service: Callable[[Session], AppTriggerPublicationService] def build_app_services( @@ -65,6 +70,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( @@ -93,7 +99,12 @@ def build_app_services( workflow_runs=DifyAPIRepositoryFactory.create_api_workflow_run_repository(session_maker=database_client), now=naive_utc_now, ), - triggers=SQLAlchemyAppTriggerRepository(session_factory=database_client), + 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/repositories/trigger/app_repository.py b/api/repositories/trigger/app_repository.py index 70872c57e41..1895bb0fd80 100644 --- a/api/repositories/trigger/app_repository.py +++ b/api/repositories/trigger/app_repository.py @@ -1,8 +1,9 @@ """App trigger management with workspace-scoped, bounded database transactions.""" +from collections.abc import Sequence from typing import override -from sqlalchemy import select +from sqlalchemy import select, update from sqlalchemy.orm import Session, sessionmaker from machinery.context import RequestContext @@ -15,6 +16,7 @@ from services.trigger.app_trigger_store import ( AppTriggerNotFoundError, AppTriggerRecord, AppTriggerStore, + PublishedAppTrigger, WebhookTriggerNotFoundError, WebhookTriggerRecord, ) @@ -90,3 +92,48 @@ class SQLAlchemyAppTriggerRepository(AppTriggerStore): 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 723d29e947f..af840f75620 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.""" + """Keep status persistence failures from replacing the original quota rejection.""" - @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. + def __init__(self, *, triggers: AppTriggerStatusWriter) -> None: + self._triggers = triggers - 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. - - 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 index 3d13994fdd6..3e680ad9bb3 100644 --- a/api/services/trigger/app_trigger_store.py +++ b/api/services/trigger/app_trigger_store.py @@ -1,5 +1,6 @@ """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 @@ -45,3 +46,21 @@ class AppTriggerStore(Protocol): 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 b3f13fd0ba3..b7685099e25 100644 --- a/api/services/trigger/webhook_service.py +++ b/api/services/trigger/webhook_service.py @@ -828,6 +828,7 @@ class WebhookService: workflow: Workflow, *, end_users: WebhookEndUserProvisioner, + trigger_status: AppTriggerService, ) -> None: """Trigger workflow execution via AsyncWorkflowService. @@ -862,7 +863,7 @@ class WebhookService: 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 2447b32136b..2c0297fbf62 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 36ed189c7c1..bc8192a6eea 100644 --- a/api/tasks/workflow_schedule_tasks.py +++ b/api/tasks/workflow_schedule_tasks.py @@ -9,11 +9,11 @@ from core.workflow.nodes.trigger_schedule.exc import ( 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/e2e_tests/containers/services/test_trigger_binding_publication.py b/api/tests/e2e_tests/containers/services/test_trigger_binding_publication.py index 2aa25b8034b..d6b3f5488f5 100644 --- a/api/tests/e2e_tests/containers/services/test_trigger_binding_publication.py +++ b/api/tests/e2e_tests/containers/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/e2e_tests/containers/services/test_webhook_service_relationships.py b/api/tests/e2e_tests/containers/services/test_webhook_service_relationships.py index ad2e9d62fd2..ef24e66bad0 100644 --- a/api/tests/e2e_tests/containers/services/test_webhook_service_relationships.py +++ b/api/tests/e2e_tests/containers/services/test_webhook_service_relationships.py @@ -13,6 +13,7 @@ from sqlalchemy.orm import Session 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 @@ -374,6 +375,7 @@ class TestWebhookServiceTriggerExecutionWithContainers: webhook_data, workflow, end_users=_EndUserServiceStub(end_user), + trigger_status=application_services().apps.trigger_status, ) mock_reserve.assert_called_once() @@ -402,14 +404,12 @@ class TestWebhookServiceTriggerExecutionWithContainers: 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( @@ -417,9 +417,11 @@ class TestWebhookServiceTriggerExecutionWithContainers: {"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, @@ -445,6 +447,7 @@ class TestWebhookServiceTriggerExecutionWithContainers: {"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/trigger/test_webhook.py b/api/tests/unit_tests/controllers/trigger/test_webhook.py index 8537780e1e2..84af766fe0d 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 dataclasses import dataclass 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.helpers.unit.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 00000000000..8ca39693d93 --- /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/repositories/trigger/test_app_repository.py b/api/tests/unit_tests/repositories/trigger/test_app_repository.py index bf32f645c17..95cb67da928 100644 --- a/api/tests/unit_tests/repositories/trigger/test_app_repository.py +++ b/api/tests/unit_tests/repositories/trigger/test_app_repository.py @@ -325,3 +325,65 @@ def test_failed_status_write_rolls_back_and_releases_connection( 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/trigger/test_webhook_service.py b/api/tests/unit_tests/services/trigger/test_webhook_service.py index b67dc0e8686..be1a6044cc1 100644 --- a/api/tests/unit_tests/services/trigger/test_webhook_service.py +++ b/api/tests/unit_tests/services/trigger/test_webhook_service.py @@ -26,14 +26,21 @@ from models.model import App, AppMode, EndUser 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 from services.trigger import webhook_service as service_module 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.helpers.unit.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 @@ -248,6 +255,7 @@ class TestWebhookServiceCases: caplog: pytest.LogCaptureFixture, monkeypatch: pytest.MonkeyPatch, sqlite_engine: _webhook_service_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)) @@ -270,6 +278,7 @@ class TestWebhookServiceCases: {"body": {}, "headers": {}, "query_params": {}, "files": {}, "method": "POST"}, workflow, end_users=_EndUserServiceStub(_end_user()), + trigger_status=trigger_status, ) assert exc_info.value is quota_error @@ -287,6 +296,7 @@ class TestWebhookServiceCases: self, monkeypatch: pytest.MonkeyPatch, sqlite_engine: _webhook_service_Engine, + trigger_status: AppTriggerService, ) -> None: monkeypatch.setattr(webhook_service_module, "db", SimpleNamespace(engine=sqlite_engine)) webhook_trigger = _webhook_trigger() @@ -313,6 +323,7 @@ class TestWebhookServiceCases: webhook_data, workflow, end_users=_EndUserServiceStub(end_user), + trigger_status=trigger_status, ) call_session = mock_trigger.call_args.kwargs["session"] @@ -320,7 +331,7 @@ class TestWebhookServiceCases: 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": {}} @@ -331,6 +342,7 @@ class TestWebhookServiceCases: 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 a14ee0ac2ec..ebcae5a8e6e 100644 --- a/api/tests/unit_tests/tasks/test_trigger_processing_tasks.py +++ b/api/tests/unit_tests/tasks/test_trigger_processing_tasks.py @@ -132,6 +132,7 @@ class _Dispatch: 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)