From 93391a421d397a2bb4e018dbbc9a476f62518ec9 Mon Sep 17 00:00:00 2001 From: hjlarry Date: Sat, 31 Oct 2026 16:17:52 +0800 Subject: [PATCH] refactor(api): inject trigger workflow dispatch and execution logging --- api/.importlinter | 2 + api/controllers/trigger/webhook.py | 1 + .../application_services/workspace.py | 36 +- api/extensions/ext_application_services.py | 6 + .../app_definition_query_repository.py | 59 ++ ...alchemy_workflow_trigger_log_repository.py | 136 ++- api/services/async_workflow_service.py | 417 ++------- api/services/trigger/webhook_service.py | 14 +- .../trigger/workflow_dispatch_gateway.py | 21 + .../trigger/workflow_dispatch_service.py | 307 +++++++ api/services/workflow/entities.py | 23 +- api/services/workflow/queue_dispatcher.py | 24 + api/tasks/trigger_processing_tasks.py | 397 +-------- api/tasks/workflow_schedule_tasks.py | 24 +- .../test_webhook_service_relationships.py | 14 +- .../trigger/test_trigger_e2e.py | 17 +- .../inner_api/workspace/test_workspace.py | 40 + .../controllers/trigger/test_webhook.py | 6 + .../test_trigger_execution_logs.py | 83 ++ .../services/test_async_workflow_service.py | 816 +++++------------- .../services/test_webhook_service.py | 101 +-- .../tasks/test_trigger_processing_tasks.py | 19 +- 22 files changed, 1112 insertions(+), 1451 deletions(-) create mode 100644 api/services/trigger/workflow_dispatch_gateway.py create mode 100644 api/services/trigger/workflow_dispatch_service.py create mode 100644 api/tests/unit_tests/repositories/test_trigger_execution_logs.py diff --git a/api/.importlinter b/api/.importlinter index bc880da786a821..3f465462f28166 100644 --- a/api/.importlinter +++ b/api/.importlinter @@ -1224,6 +1224,8 @@ source_modules = services.trigger.subscription_renewal_service services.trigger.subscription_refresh_service services.trigger.subscription_binding_service + services.async_workflow_service + services.trigger.workflow_dispatch_service forbidden_modules = controllers extensions diff --git a/api/controllers/trigger/webhook.py b/api/controllers/trigger/webhook.py index a2f46418de2eaa..57caaafc81dbfc 100644 --- a/api/controllers/trigger/webhook.py +++ b/api/controllers/trigger/webhook.py @@ -78,6 +78,7 @@ def handle_webhook(webhook_id: str): workflow, end_users=application_services().app_scoped_end_users.commands, trigger_status=application_services().apps.trigger_status, + workflow_dispatch=application_services().workspaces.workflow_dispatch, ) # Return configured response diff --git a/api/extensions/application_services/workspace.py b/api/extensions/application_services/workspace.py index 77d5ddb49b5b1c..dbf4cd0ab81b51 100644 --- a/api/extensions/application_services/workspace.py +++ b/api/extensions/application_services/workspace.py @@ -14,6 +14,7 @@ from repositories.app.console_repository import ConsoleAppRepository from repositories.credentials.query_repository import CredentialQueryRepository from repositories.knowledge.dataset_repository import SQLAlchemyDatasetRepository +from repositories.sqlalchemy_workflow_trigger_log_repository import SQLAlchemyWorkflowTriggerDispatchRepository from repositories.trigger.oauth_client_repository import TriggerOAuthClientRepository from repositories.trigger.subscription_builder_repository import RedisTriggerSubscriptionBuilderRepository from repositories.trigger.subscription_repository import ( @@ -24,9 +25,11 @@ from repositories.workspace.resource_cleanup_repository import WorkspaceResourceCleanupRepository from repositories.workspace.workspace_repository import WorkspaceRepository from services.account.adapters import RedisInvitationTokenStore +from services.async_workflow_service import AsyncWorkflowService, TriggerWorkflowQuery from services.file_service import FileService from services.plugin.plugin_parameter_gateway import DaemonPluginParameterGateway from services.plugin.plugin_parameter_service import PluginParameterService +from services.quota_service import QuotaService from services.trigger.endpoint_gateway import PluginTriggerEndpointGateway, enqueue_triggered_workflows from services.trigger.endpoint_service import TriggerEndpointService from services.trigger.event_gateway import PluginTriggerEventGateway @@ -55,6 +58,8 @@ from services.trigger.subscription_service import TriggerSubscriptionService from services.trigger.trigger_request_service import TriggerHttpRequestCachingService from services.trigger.trigger_subscription_builder_service import TriggerSubscriptionBuilderService +from services.trigger.workflow_dispatch_service import PluginTriggerWorkflowDispatchService, TriggerEndUserProvisioner +from services.workflow.queue_dispatcher import CeleryWorkflowQueuePublisher from services.workspace.gateways import ( DeploymentWorkspaceCreationPolicy, DeploymentWorkspaceFeatureGateway, @@ -101,6 +106,8 @@ class WorkspaceServices: trigger_builders: TriggerSubscriptionBuilderService trigger_endpoints: TriggerEndpointService trigger_events: TriggerEventService + workflow_dispatch: AsyncWorkflowService + trigger_workflow_dispatch: PluginTriggerWorkflowDispatchService trigger_oauth_clients: TriggerOAuthClientService trigger_oauth: TriggerOAuthService @@ -139,6 +146,9 @@ def build_workspace_services( registration: InvitedAccountRegistration, invitation_tokens: RedisInvitationTokenStore, credentials: CredentialQueryRepository, + workflow_queries: TriggerWorkflowQuery, + trigger_end_users: TriggerEndUserProvisioner, + mark_rate_limited: Callable[[str], None], ) -> WorkspaceServices: trigger_providers = PluginTriggerProviderQuery() subscriptions = TriggerSubscriptionRepository(session_factory=database_client) @@ -171,6 +181,16 @@ def build_workspace_services( invalidate_cache=delete_cache_for_subscription, invalidate_properties=invalidate_subscription_properties, ) + trigger_events = TriggerEventService( + subscriptions=trigger_subscriptions, + plugins=PluginTriggerEventGateway(), + requests=TriggerHttpRequestCachingService, + ) + execution_logs = SQLAlchemyWorkflowTriggerDispatchRepository(session_factory=database_client) + queues = CeleryWorkflowQueuePublisher() + workflow_dispatch = AsyncWorkflowService( + workflows=workflow_queries, logs=execution_logs, queues=queues, reserve_quota=QuotaService.reserve + ) return WorkspaceServices( queries=WorkspaceQueryService(workspaces=workspaces, plans=DeploymentWorkspacePlanGateway()), management=WorkspaceService( @@ -216,10 +236,18 @@ def build_workspace_services( create_trigger_binding_service=lambda session: TriggerSubscriptionBindingService( sync_bindings=partial(subscriptions.sync_workflow_bindings, session=session) ), - trigger_events=TriggerEventService( - subscriptions=trigger_subscriptions, - plugins=PluginTriggerEventGateway(), - requests=TriggerHttpRequestCachingService, + trigger_events=trigger_events, + workflow_dispatch=workflow_dispatch, + trigger_workflow_dispatch=PluginTriggerWorkflowDispatchService( + workflows=workflow_queries, + logs=execution_logs, + queues=queues, + dispatch=workflow_dispatch, + events=trigger_events, + bindings=subscriptions, + end_users=trigger_end_users, + reserve_quota=QuotaService.reserve, + mark_rate_limited=mark_rate_limited, ), trigger_subscription_management=trigger_subscription_management, trigger_subscription_renewal=renewal, diff --git a/api/extensions/ext_application_services.py b/api/extensions/ext_application_services.py index 18a6a15d77b6cc..e0fb5e4e6c0eb5 100644 --- a/api/extensions/ext_application_services.py +++ b/api/extensions/ext_application_services.py @@ -187,6 +187,7 @@ from services.system_feature_service import SystemFeatureService from services.tag_application_service import TagApplicationService from services.tool_file_download_service import ToolFileDownloadService +from services.trigger.workflow_dispatch_gateway import AppScopedTriggerEndUsers from services.upload_file_delivery_service import UploadFileDeliveryService from services.web_app_runtime_query_service import WebAppRuntimeQueryService from services.web_authentication_adapters import ( @@ -523,6 +524,11 @@ def build_application_services( registration=account_services.lifecycle, invitation_tokens=invitation_tokens, credentials=credential_queries, + workflow_queries=app_definition_repository, + mark_rate_limited=apps.trigger_status.mark_tenant_triggers_rate_limited, + trigger_end_users=AppScopedTriggerEndUsers( + end_users=AppScopedEndUserService(end_users=app_scoped_end_user_repository) + ), ) return ApplicationServices( rbac=rbac, diff --git a/api/repositories/app_definition_query_repository.py b/api/repositories/app_definition_query_repository.py index 0fae049be293db..eef4fdaf4d7d09 100644 --- a/api/repositories/app_definition_query_repository.py +++ b/api/repositories/app_definition_query_repository.py @@ -1,5 +1,6 @@ """Database repository for externally visible app definitions.""" +from collections.abc import Mapping, Sequence from typing import Any, cast, override from sqlalchemy import select @@ -22,7 +23,9 @@ AppSiteConfiguration, AppToolIconSource, ) +from services.errors.app import IsDraftWorkflowError, WorkflowNotFoundError from services.web_app_runtime_query_service import WebAppRuntimeRecord +from services.workflow.entities import TriggerWorkflowSnapshot def map_site_configuration(site: Site) -> AppSiteConfiguration: @@ -118,6 +121,62 @@ class AppDefinitionQueryRepository(AppDefinitionQuery): def __init__(self, *, session_factory: sessionmaker[Session]) -> None: self._session_factory = session_factory + def get_trigger_workflow(self, *, tenant_id: str, app_id: str, workflow_id: str | None) -> TriggerWorkflowSnapshot: + with self._session_factory() as session: + app = session.scalar(select(App).where(App.id == app_id, App.tenant_id == tenant_id)) + if app is None: + raise WorkflowNotFoundError(f"App not found: {app_id}") + selected_id = workflow_id or app.workflow_id + workflow = ( + session.scalar( + select(Workflow).where( + Workflow.id == selected_id, Workflow.app_id == app_id, Workflow.tenant_id == tenant_id + ) + ) + if selected_id + else None + ) + if workflow is None: + if workflow_id: + raise WorkflowNotFoundError(f"Published workflow not found: {workflow_id}") + raise WorkflowNotFoundError(f"No published workflow found for app: {app_id}") + if workflow_id and workflow.version == Workflow.VERSION_DRAFT: + raise IsDraftWorkflowError( + f"Cannot use draft workflow version. Workflow ID: {workflow_id}. " + "Please use a published workflow version or leave workflow_id empty." + ) + return self._trigger_workflow_snapshot(workflow) + + def get_published_trigger_workflows( + self, *, tenant_id: str, app_ids: Sequence[str] + ) -> Mapping[str, TriggerWorkflowSnapshot]: + if not app_ids: + return {} + with self._session_factory() as session: + workflows = session.scalars( + select(Workflow) + .join(App, App.workflow_id == Workflow.id) + .where( + App.id.in_(app_ids), + App.tenant_id == tenant_id, + Workflow.app_id == App.id, + Workflow.tenant_id == App.tenant_id, + Workflow.version != Workflow.VERSION_DRAFT, + ) + ).all() + return {workflow.app_id: self._trigger_workflow_snapshot(workflow) for workflow in workflows} + + @staticmethod + def _trigger_workflow_snapshot(workflow: Workflow) -> TriggerWorkflowSnapshot: + return TriggerWorkflowSnapshot( + id=workflow.id, + tenant_id=workflow.tenant_id, + app_id=workflow.app_id, + type=workflow.type, + version=workflow.version, + graph=workflow.graph, + ) + @override def get_mode(self, app_id: str) -> str | None: with self._session_factory() as session: diff --git a/api/repositories/sqlalchemy_workflow_trigger_log_repository.py b/api/repositories/sqlalchemy_workflow_trigger_log_repository.py index f805fa0e4b1c6f..bbf11fd13a34af 100644 --- a/api/repositories/sqlalchemy_workflow_trigger_log_repository.py +++ b/api/repositories/sqlalchemy_workflow_trigger_log_repository.py @@ -2,17 +2,21 @@ SQLAlchemy implementation of WorkflowTriggerLogRepository. """ -from collections.abc import Sequence +import json +from collections.abc import Mapping, Sequence from datetime import UTC, datetime, timedelta from typing import cast, override -from sqlalchemy import and_, delete, func, select +from sqlalchemy import and_, delete, func, select, update from sqlalchemy.engine import CursorResult -from sqlalchemy.orm import Session +from sqlalchemy.orm import Session, sessionmaker -from models.enums import WorkflowTriggerStatus +from graphon.enums import WorkflowExecutionStatus +from models.enums import WorkflowRunTriggeredFrom, WorkflowTriggerStatus from models.trigger import WorkflowTriggerLog +from models.workflow import WorkflowAppLog, WorkflowAppLogCreatedFrom, WorkflowRun from repositories.workflow_trigger_log_repository import WorkflowTriggerLogRepository +from services.workflow.entities import TriggerActor, TriggerData, TriggerWorkflowSnapshot class SQLAlchemyWorkflowTriggerLogRepository(WorkflowTriggerLogRepository): @@ -142,3 +146,127 @@ def count_by_run_ids(self, run_ids: Sequence[str]) -> int: select(func.count()).select_from(WorkflowTriggerLog).where(WorkflowTriggerLog.workflow_run_id.in_(run_ids)) ) return int(count or 0) + + +class SQLAlchemyWorkflowTriggerDispatchRepository: + """Transaction-owning trigger log writes for producers. + + Runtime and retention code still borrow a caller session through the repository + above. These producer operations close their sessions before broker or billing I/O. + """ + + def __init__(self, *, session_factory: sessionmaker[Session]) -> None: + self._session_factory = session_factory + + def create_pending(self, *, actor: TriggerActor, trigger_data: TriggerData, workflow_id: str, queue: str) -> str: + log = self._new_log(actor=actor, trigger_data=trigger_data, workflow_id=workflow_id, queue=queue) + with self._session_factory.begin() as session: + SQLAlchemyWorkflowTriggerLogRepository(session).create(log) + log_id = log.id + return log_id + + def mark_failed( + self, *, tenant_id: str, app_id: str, log_id: str, status: WorkflowTriggerStatus, error: str + ) -> None: + with self._session_factory.begin() as session: + session.execute( + update(WorkflowTriggerLog) + .where( + WorkflowTriggerLog.id == log_id, + WorkflowTriggerLog.tenant_id == tenant_id, + WorkflowTriggerLog.app_id == app_id, + ) + .values(status=status, error=error) + ) + + def mark_queued(self, *, tenant_id: str, app_id: str, log_id: str, task_id: str) -> None: + with self._session_factory.begin() as session: + session.execute( + update(WorkflowTriggerLog) + .where( + WorkflowTriggerLog.id == log_id, + WorkflowTriggerLog.tenant_id == tenant_id, + WorkflowTriggerLog.app_id == app_id, + ) + .values(status=WorkflowTriggerStatus.QUEUED, celery_task_id=task_id, triggered_at=datetime.now(UTC)) + ) + + def record_invocation_failure( + self, + *, + workflow: TriggerWorkflowSnapshot, + actor: TriggerActor, + trigger_data: TriggerData, + failure_inputs: Mapping[str, str], + error: str, + queue: str, + ) -> None: + now = datetime.now(UTC) + run = WorkflowRun( + tenant_id=workflow.tenant_id, + app_id=workflow.app_id, + workflow_id=workflow.id, + type=workflow.type, + triggered_from=WorkflowRunTriggeredFrom.PLUGIN.value, + version=workflow.version, + graph=workflow.graph, + inputs=json.dumps(dict(failure_inputs)), + status=WorkflowExecutionStatus.FAILED.value, + outputs="{}", + error=error, + elapsed_time=0.0, + total_tokens=0, + total_steps=0, + created_by_role=actor.role, + created_by=actor.id, + created_at=now, + finished_at=now, + exceptions_count=0, + ) + log = self._new_log(actor=actor, trigger_data=trigger_data, workflow_id=workflow.id, queue=queue) + log.status = WorkflowTriggerStatus.FAILED + log.error = error + log.triggered_at = now + log.finished_at = now + log.elapsed_time = 0.0 + log.total_tokens = 0 + with self._session_factory.begin() as session: + session.add(run) + session.flush() + session.add( + WorkflowAppLog( + tenant_id=workflow.tenant_id, + app_id=workflow.app_id, + workflow_id=workflow.id, + workflow_run_id=run.id, + created_from=WorkflowAppLogCreatedFrom.SERVICE_API, + created_by_role=actor.role, + created_by=actor.id, + ) + ) + log.workflow_run_id = run.id + SQLAlchemyWorkflowTriggerLogRepository(session).create(log) + + @staticmethod + def _new_log(*, actor: TriggerActor, trigger_data: TriggerData, workflow_id: str, queue: str) -> WorkflowTriggerLog: + return WorkflowTriggerLog( + tenant_id=trigger_data.tenant_id, + app_id=trigger_data.app_id, + workflow_id=workflow_id, + root_node_id=trigger_data.root_node_id, + trigger_metadata=trigger_data.trigger_metadata.model_dump_json() if trigger_data.trigger_metadata else "{}", + trigger_type=trigger_data.trigger_type, + workflow_run_id=None, + outputs=None, + trigger_data=trigger_data.model_dump_json(), + inputs=json.dumps(dict(trigger_data.inputs)), + status=WorkflowTriggerStatus.PENDING, + queue_name=queue, + retry_count=0, + created_by_role=actor.role, + created_by=actor.id, + celery_task_id=None, + error=None, + elapsed_time=None, + total_tokens=None, + ) diff --git a/api/services/async_workflow_service.py b/api/services/async_workflow_service.py index aaf01b8c2398d3..872ed1012ff06a 100644 --- a/api/services/async_workflow_service.py +++ b/api/services/async_workflow_service.py @@ -1,363 +1,128 @@ -""" -Universal async workflow execution service. +"""Queue trigger executions using detached inputs and bounded persistence operations.""" -This service provides a centralized entry point for triggering workflows asynchronously -with support for different subscription tiers, rate limiting, and execution tracking. -""" - -import json import logging -from datetime import UTC, datetime -from typing import Any - -from celery.result import AsyncResult -from sqlalchemy import select -from sqlalchemy.orm import Session, sessionmaker +from collections.abc import Callable, Mapping, Sequence +from typing import Protocol from enums import QuotaType -from extensions.ext_database import db -from models.account import Account -from models.enums import CreatorUserRole, WorkflowTriggerStatus -from models.model import App, EndUser -from models.trigger import WorkflowTriggerLog, WorkflowTriggerLogDict -from models.workflow import Workflow -from repositories.sqlalchemy_workflow_trigger_log_repository import SQLAlchemyWorkflowTriggerLogRepository -from services.errors.app import QuotaExceededError, WorkflowNotFoundError -from services.quota_service import QuotaService, unlimited -from services.workflow.entities import AsyncTriggerResponse, TriggerData, WorkflowTaskData -from services.workflow.queue_dispatcher import QueueDispatcherManager, QueuePriority -from services.workflow_service import WorkflowService -from tasks.async_workflow_tasks import ( - execute_workflow_professional, - execute_workflow_sandbox, - execute_workflow_team, -) +from models.enums import WorkflowTriggerStatus +from services.errors.app import QuotaExceededError +from services.quota_service import QuotaCharge +from services.workflow.entities import AsyncTriggerResponse, TriggerActor, TriggerData, TriggerWorkflowSnapshot logger = logging.getLogger(__name__) -class AsyncWorkflowService: - """ - Universal entry point for async workflow execution - ALL METHODS ARE NON-BLOCKING +class TriggerWorkflowQuery(Protocol): + def get_trigger_workflow( + self, *, tenant_id: str, app_id: str, workflow_id: str | None + ) -> TriggerWorkflowSnapshot: ... - This service handles: - - Trigger data validation and processing - - Queue routing based on subscription tier - - Daily rate limiting with timezone support - - Execution tracking and logging - - Retry mechanisms for failed executions + def get_published_trigger_workflows( + self, *, tenant_id: str, app_ids: Sequence[str] + ) -> Mapping[str, TriggerWorkflowSnapshot]: ... - Important: All trigger methods return immediately after queuing tasks. - Actual workflow execution happens asynchronously in background Celery workers. - Use trigger log IDs to monitor execution status and results. - """ - @classmethod - def trigger_workflow_async( - cls, user: Account | EndUser, trigger_data: TriggerData, *, session: Session - ) -> AsyncTriggerResponse: - """ - Universal entry point for async workflow execution - THIS METHOD WILL NOT BLOCK +class TriggerExecutionLogs(Protocol): + def create_pending( + self, *, actor: TriggerActor, trigger_data: TriggerData, workflow_id: str, queue: str + ) -> str: ... - Creates a trigger log and dispatches to appropriate queue based on subscription tier. - The workflow execution happens asynchronously in the background via Celery workers. - This method returns immediately after queuing the task, not after execution completion. + def mark_failed( + self, *, tenant_id: str, app_id: str, log_id: str, status: WorkflowTriggerStatus, error: str + ) -> None: ... - Args: - session: Database session to use for operations - user: User (Account or EndUser) who initiated the workflow trigger - trigger_data: Validated Pydantic model containing trigger information + def mark_queued(self, *, tenant_id: str, app_id: str, log_id: str, task_id: str) -> None: ... - Returns: - AsyncTriggerResponse with workflow_trigger_log_id, task_id, status="queued", and queue - Note: The actual workflow execution status must be checked separately via workflow_trigger_log_id + def record_invocation_failure( + self, + *, + workflow: TriggerWorkflowSnapshot, + actor: TriggerActor, + trigger_data: TriggerData, + failure_inputs: Mapping[str, str], + error: str, + queue: str, + ) -> None: ... - Raises: - WorkflowNotFoundError: If app or workflow not found - QuotaExceededError: If workflow execution quota is exhausted - Behavior: - - Non-blocking: Returns immediately after queuing - - Asynchronous: Actual execution happens in background Celery workers - - Status tracking: Use workflow_trigger_log_id to monitor progress - - Queue-based: Routes to different queues based on subscription tier - """ - trigger_log_repo = SQLAlchemyWorkflowTriggerLogRepository(session) - dispatcher_manager = QueueDispatcherManager() - workflow_service = WorkflowService() +class TriggerWorkflowDispatcher(Protocol): + def trigger_workflow_async(self, *, actor: TriggerActor, trigger_data: TriggerData) -> AsyncTriggerResponse: ... - # 1. Validate app exists - app_model = session.scalar(select(App).where(App.id == trigger_data.app_id)) - if not app_model: - raise WorkflowNotFoundError(f"App not found: {trigger_data.app_id}") - # 2. Get workflow - workflow = cls._get_workflow(workflow_service, app_model, trigger_data.workflow_id, session=session) +class WorkflowQueuePublisher(Protocol): + def get_queue(self, tenant_id: str) -> str: ... - # commit read only session before starting the billig rpc call - session.commit() + def publish(self, *, queue: str, trigger_log_id: str) -> str: ... - # 3. Get dispatcher based on tenant subscription - dispatcher = dispatcher_manager.get_dispatcher(trigger_data.tenant_id) - # 4. Rate limiting check will be done without timezone first +class AsyncWorkflowService: + """Create execution logs and enqueue tasks without retaining a database session. - # 5. Determine user role and ID - if isinstance(user, Account): - created_by_role = CreatorUserRole.ACCOUNT - created_by = user.id - else: # EndUser - created_by_role = CreatorUserRole.END_USER - created_by = user.id + Log creation, quota reservation, broker publication and final status are distinct + steps, as before. This is not an outbox: broker publication can still succeed + before recording its task ID fails. + """ - # 6. Create trigger log entry first (for tracking) - trigger_log = WorkflowTriggerLog( - tenant_id=trigger_data.tenant_id, - app_id=trigger_data.app_id, - workflow_id=workflow.id, - root_node_id=trigger_data.root_node_id, - trigger_metadata=( - trigger_data.trigger_metadata.model_dump_json() if trigger_data.trigger_metadata else "{}" - ), - trigger_type=trigger_data.trigger_type, - workflow_run_id=None, - outputs=None, - trigger_data=trigger_data.model_dump_json(), - inputs=json.dumps(dict(trigger_data.inputs)), - status=WorkflowTriggerStatus.PENDING, - queue_name=dispatcher.get_queue_name(), - retry_count=0, - created_by_role=created_by_role, - created_by=created_by, - celery_task_id=None, - error=None, - elapsed_time=None, - total_tokens=None, + def __init__( + self, + *, + workflows: TriggerWorkflowQuery, + logs: TriggerExecutionLogs, + queues: WorkflowQueuePublisher, + reserve_quota: Callable[[QuotaType, str], QuotaCharge], + ) -> None: + self._workflows = workflows + self._logs = logs + self._queues = queues + self._reserve_quota = reserve_quota + + def trigger_workflow_async(self, *, actor: TriggerActor, trigger_data: TriggerData) -> AsyncTriggerResponse: + workflow = self._workflows.get_trigger_workflow( + tenant_id=trigger_data.tenant_id, app_id=trigger_data.app_id, workflow_id=trigger_data.workflow_id ) - - trigger_log = trigger_log_repo.create(trigger_log) - session.commit() - - # 7. Reserve quota (commit after successful dispatch) - quota_charge = unlimited() + queue = self._queues.get_queue(trigger_data.tenant_id) + log_id = self._logs.create_pending(actor=actor, trigger_data=trigger_data, workflow_id=workflow.id, queue=queue) try: - quota_charge = QuotaService.reserve(QuotaType.WORKFLOW, trigger_data.tenant_id) - except QuotaExceededError as e: - # Update trigger log status - trigger_log.status = WorkflowTriggerStatus.RATE_LIMITED - trigger_log.error = f"Quota limit reached: {e}" - trigger_log_repo.update(trigger_log) - session.commit() + charge = self._reserve_quota(QuotaType.WORKFLOW, trigger_data.tenant_id) + except QuotaExceededError as exc: + self._logs.mark_failed( + tenant_id=trigger_data.tenant_id, + app_id=trigger_data.app_id, + log_id=log_id, + status=WorkflowTriggerStatus.RATE_LIMITED, + error=f"Quota limit reached: {exc}", + ) logger.info( - "Workflow quota exceeded for tenant %s, app %s, workflow %s, trigger log %s", + "Workflow quota exceeded for tenant %s, app %s, trigger log %s", trigger_data.tenant_id, trigger_data.app_id, - workflow.id, - trigger_log.id, + log_id, ) - raise - - # 8. Create task data - queue_name = dispatcher.get_queue_name() - - task_data = WorkflowTaskData(workflow_trigger_log_id=trigger_log.id) - - # 9. Dispatch to appropriate queue - task_data_dict = task_data.model_dump(mode="json") - try: - task: AsyncResult[Any] | None = None - if queue_name == QueuePriority.PROFESSIONAL: - task = execute_workflow_professional.delay(task_data_dict) - elif queue_name == QueuePriority.TEAM: - task = execute_workflow_team.delay(task_data_dict) - else: # SANDBOX - task = execute_workflow_sandbox.delay(task_data_dict) - quota_charge.commit() - except Exception as e: - # Mark the trigger log as FAILED so it does not stay PENDING when the broker - # or dispatcher refuses the task, then re-raise so the caller still surfaces - # the error. - trigger_log.status = WorkflowTriggerStatus.FAILED - trigger_log.error = f"Failed to dispatch workflow task: {e}" - trigger_log_repo.update(trigger_log) - session.commit() - quota_charge.refund() + task_id = self._queues.publish(queue=queue, trigger_log_id=log_id) + charge.commit() + except Exception as exc: + # Refund even when recording the publication failure also fails. + try: + self._logs.mark_failed( + tenant_id=trigger_data.tenant_id, + app_id=trigger_data.app_id, + log_id=log_id, + status=WorkflowTriggerStatus.FAILED, + error=f"Failed to dispatch workflow task: {exc}", + ) + finally: + charge.refund() logger.exception( - "Failed to dispatch workflow task for tenant %s, app %s, workflow %s, trigger log %s", + "Failed to dispatch workflow task for tenant %s, app %s, trigger log %s", trigger_data.tenant_id, trigger_data.app_id, - workflow.id, - trigger_log.id, + log_id, ) raise - - # 10. Update trigger log with task info - trigger_log.status = WorkflowTriggerStatus.QUEUED - trigger_log.celery_task_id = task.id - trigger_log.triggered_at = datetime.now(UTC) - trigger_log_repo.update(trigger_log) - session.commit() - - return AsyncTriggerResponse( - workflow_trigger_log_id=trigger_log.id, - task_id=task.id, - status="queued", - queue=queue_name, + self._logs.mark_queued( + tenant_id=trigger_data.tenant_id, app_id=trigger_data.app_id, log_id=log_id, task_id=task_id ) - - @classmethod - def reinvoke_trigger( - cls, user: Account | EndUser, workflow_trigger_log_id: str, *, session: Session - ) -> AsyncTriggerResponse: - """ - Re-invoke a previously failed or rate-limited trigger - THIS METHOD WILL NOT BLOCK - - Updates the existing trigger log to retry status and creates a new async execution. - Returns immediately after queuing the retry, not after execution completion. - - Args: - session: Database session to use for operations - user: User (Account or EndUser) who initiated the retry - workflow_trigger_log_id: ID of the trigger log to re-invoke - - Returns: - AsyncTriggerResponse with new execution information (status="queued") - Note: This creates a new trigger log entry for the retry attempt - - Raises: - ValueError: If trigger log not found - QuotaExceededError: If workflow execution quota is exhausted - - Behavior: - - Non-blocking: Returns immediately after queuing retry - - Creates new trigger log: Original log marked as retrying, new log for execution - - Preserves original trigger data: Uses same inputs and configuration - """ - trigger_log_repo = SQLAlchemyWorkflowTriggerLogRepository(session) - - trigger_log = trigger_log_repo.get_by_id(workflow_trigger_log_id) - - if not trigger_log: - raise ValueError(f"Trigger log not found: {workflow_trigger_log_id}") - - # Reconstruct trigger data from log - trigger_data = TriggerData.model_validate_json(trigger_log.trigger_data) - - # Reset log for retry - trigger_log.status = WorkflowTriggerStatus.RETRYING - trigger_log.retry_count += 1 - trigger_log.error = None - trigger_log.triggered_at = datetime.now(UTC) - trigger_log_repo.update(trigger_log) - session.commit() - - # Re-trigger workflow (this will create a new trigger log) - return cls.trigger_workflow_async(user, trigger_data, session=session) - - @classmethod - def get_trigger_log( - cls, workflow_trigger_log_id: str, tenant_id: str | None = None - ) -> WorkflowTriggerLogDict | None: - """ - Get trigger log by ID - - Args: - workflow_trigger_log_id: ID of the trigger log - tenant_id: Optional tenant ID for security check - - Returns: - Trigger log as dictionary or None if not found - """ - with sessionmaker(db.engine).begin() as session: - trigger_log_repo = SQLAlchemyWorkflowTriggerLogRepository(session) - trigger_log = trigger_log_repo.get_by_id(workflow_trigger_log_id, tenant_id) - - if not trigger_log: - return None - - return trigger_log.to_dict() - - @classmethod - def get_recent_logs( - cls, tenant_id: str, app_id: str, hours: int = 24, limit: int = 100, offset: int = 0 - ) -> list[WorkflowTriggerLogDict]: - """ - Get recent trigger logs - - Args: - tenant_id: Tenant ID - app_id: Application ID - hours: Number of hours to look back - limit: Maximum number of results - offset: Number of results to skip - - Returns: - List of trigger logs as dictionaries - """ - with sessionmaker(db.engine).begin() as session: - trigger_log_repo = SQLAlchemyWorkflowTriggerLogRepository(session) - logs = trigger_log_repo.get_recent_logs( - tenant_id=tenant_id, app_id=app_id, hours=hours, limit=limit, offset=offset - ) - - return [log.to_dict() for log in logs] - - @classmethod - def get_failed_logs_for_retry( - cls, tenant_id: str, max_retry_count: int = 3, limit: int = 100 - ) -> list[WorkflowTriggerLogDict]: - """ - Get failed logs eligible for retry - - Args: - tenant_id: Tenant ID - max_retry_count: Maximum retry count - limit: Maximum number of results - - Returns: - List of failed trigger logs as dictionaries - """ - with sessionmaker(db.engine).begin() as session: - trigger_log_repo = SQLAlchemyWorkflowTriggerLogRepository(session) - logs = trigger_log_repo.get_failed_for_retry( - tenant_id=tenant_id, max_retry_count=max_retry_count, limit=limit - ) - - return [log.to_dict() for log in logs] - - @staticmethod - def _get_workflow( - workflow_service: WorkflowService, - app_model: App, - workflow_id: str | None = None, - *, - session: Session, - ) -> Workflow: - """ - Get workflow for the app - - Args: - app_model: App model instance - workflow_id: Optional specific workflow ID - session: SQLAlchemy session used for the workflow lookup. - - Returns: - Workflow instance - - Raises: - WorkflowNotFoundError: If workflow not found - """ - if workflow_id: - # Get specific published workflow - workflow = workflow_service.get_published_workflow_by_id(app_model, workflow_id, session=session) - if not workflow: - raise WorkflowNotFoundError(f"Published workflow not found: {workflow_id}") - else: - # Get default published workflow - workflow = workflow_service.get_published_workflow(app_model, session=session) - if not workflow: - raise WorkflowNotFoundError(f"No published workflow found for app: {app_model.id}") - - return workflow + return AsyncTriggerResponse(workflow_trigger_log_id=log_id, task_id=task_id, status="queued", queue=queue) diff --git a/api/services/trigger/webhook_service.py b/api/services/trigger/webhook_service.py index 71cf399b92ebeb..c3af3fb8afac9c 100644 --- a/api/services/trigger/webhook_service.py +++ b/api/services/trigger/webhook_service.py @@ -30,15 +30,15 @@ from graphon.entities.graph_config import NodeConfigDict from graphon.file import FileTransferMethod from graphon.variables.types import ArrayValidation, SegmentType -from models.enums import AppTriggerStatus, AppTriggerType, EndUserType +from models.enums import AppTriggerStatus, AppTriggerType, CreatorUserRole, EndUserType from models.model import App, EndUser from models.trigger import AppTrigger, WorkflowWebhookTrigger from models.workflow import Workflow -from services.async_workflow_service import AsyncWorkflowService +from services.async_workflow_service import TriggerWorkflowDispatcher from services.errors.app import QuotaExceededError from services.quota_service import QuotaService from services.trigger.app_trigger_service import AppTriggerService -from services.workflow.entities import WebhookTriggerData +from services.workflow.entities import TriggerActor, WebhookTriggerData from services.workflow_service import WorkflowService try: @@ -808,6 +808,7 @@ def trigger_workflow_execution( *, end_users: WebhookEndUserProvisioner, trigger_status: AppTriggerService, + workflow_dispatch: TriggerWorkflowDispatcher, ) -> None: """Trigger workflow execution via AsyncWorkflowService. @@ -846,10 +847,9 @@ def trigger_workflow_execution( raise try: - # NOTE: don not use `with sessionmaker(bind=db.engine, expire_on_commit=False).begin()` - # trigger_workflow_async need to handle multiple session commits internally - with Session(db.engine, expire_on_commit=False) as session: - AsyncWorkflowService.trigger_workflow_async(end_user, trigger_data, session=session) + workflow_dispatch.trigger_workflow_async( + actor=TriggerActor(id=end_user.id, role=CreatorUserRole.END_USER), trigger_data=trigger_data + ) quota_charge.commit() except Exception: quota_charge.refund() diff --git a/api/services/trigger/workflow_dispatch_gateway.py b/api/services/trigger/workflow_dispatch_gateway.py new file mode 100644 index 00000000000000..2de3c15d118820 --- /dev/null +++ b/api/services/trigger/workflow_dispatch_gateway.py @@ -0,0 +1,21 @@ +"""Convert the legacy end-user provisioner's ORM results at the runtime boundary.""" + +from collections.abc import Mapping + +from models.enums import CreatorUserRole, EndUserType +from models.model import EndUser +from services.app_scoped_end_user_service import AppScopedEndUserService +from services.workflow.entities import TriggerActor + + +class AppScopedTriggerEndUsers: + def __init__(self, *, end_users: AppScopedEndUserService[EndUser]) -> None: + self._end_users = end_users + + def create_end_user_batch( + self, type: EndUserType, tenant_id: str, app_ids: list[str], user_id: str + ) -> Mapping[str, TriggerActor]: + records = self._end_users.create_end_user_batch( + type=type, tenant_id=tenant_id, app_ids=app_ids, user_id=user_id + ) + return {app_id: TriggerActor(id=record.id, role=CreatorUserRole.END_USER) for app_id, record in records.items()} diff --git a/api/services/trigger/workflow_dispatch_service.py b/api/services/trigger/workflow_dispatch_service.py new file mode 100644 index 00000000000000..5437639a863b1a --- /dev/null +++ b/api/services/trigger/workflow_dispatch_service.py @@ -0,0 +1,307 @@ +"""Dispatch subscribed plugin events into workflow executions outside SQL transactions.""" + +import json +import logging +from collections.abc import Callable, Mapping +from typing import Protocol + +from core.plugin.entities.request import TriggerInvokeEventResponse +from core.plugin.impl.exc import PluginInvokeError +from core.trigger.constants import TRIGGER_PLUGIN_NODE_TYPE +from core.workflow.nodes.trigger_plugin.entities import TriggerEventNodeData +from enums import QuotaType +from models.enums import CreatorUserRole, EndUserType +from services.async_workflow_service import ( + TriggerExecutionLogs, + TriggerWorkflowDispatcher, + TriggerWorkflowQuery, + WorkflowQueuePublisher, +) +from services.errors.app import QuotaExceededError +from services.quota_service import QuotaCharge, unlimited +from services.trigger.event_service import TriggerEventService +from services.trigger.subscription_binding_service import TriggerSubscriber, TriggerSubscriptionBindingQuery +from services.trigger.subscription_query_service import TriggerSubscriptionRuntime +from services.workflow.entities import PluginTriggerData, PluginTriggerMetadata, TriggerActor, TriggerWorkflowSnapshot + +logger = logging.getLogger(__name__) + + +class TriggerEndUserProvisioner(Protocol): + def create_end_user_batch( + self, + type: EndUserType, + tenant_id: str, + app_ids: list[str], + user_id: str, + ) -> Mapping[str, TriggerActor]: ... + + +class PluginTriggerWorkflowDispatchService: + def __init__( + self, + *, + workflows: TriggerWorkflowQuery, + logs: TriggerExecutionLogs, + queues: WorkflowQueuePublisher, + dispatch: TriggerWorkflowDispatcher, + events: TriggerEventService, + bindings: TriggerSubscriptionBindingQuery, + end_users: TriggerEndUserProvisioner, + reserve_quota: Callable[[QuotaType, str], QuotaCharge], + mark_rate_limited: Callable[[str], None], + ) -> None: + self._workflows = workflows + self._logs = logs + self._queues = queues + self._dispatch = dispatch + self._events = events + self._bindings = bindings + self._end_users = end_users + self._reserve_quota = reserve_quota + self._mark_rate_limited = mark_rate_limited + + def dispatch_event( + self, + *, + user_id: str, + subscription: TriggerSubscriptionRuntime, + event_name: str, + request_id: str, + ) -> int: + """Process triggered workflows. + + Args: + user_id: The ID of the user who activated the trigger + subscription: The trigger subscription + event_name: The name of the trigger event that was activated + request_id: The ID of the stored request in storage system + + Returns: + The number of workflows dispatched for the event + """ + subscribers = self._bindings.get_subscriber_triggers( + tenant_id=subscription.tenant_id, subscription_id=subscription.id, event_name=event_name + ) + if not subscribers: + logger.warning( + "No workflows found for trigger event '%s' in subscription '%s'", + event_name, + subscription.id, + ) + return 0 + + dispatched_count = 0 + prepared_event = self._events.prepare(subscription=subscription, event_name=event_name, request_id=request_id) + definition = prepared_event.definition + + workflows = self._workflows.get_published_trigger_workflows( + tenant_id=subscription.tenant_id, app_ids=[subscriber.app_id for subscriber in subscribers] + ) + + end_users_by_app = self._end_users.create_end_user_batch( + type=EndUserType.TRIGGER, + tenant_id=subscription.tenant_id, + app_ids=[plugin_trigger.app_id for plugin_trigger in subscribers], + user_id=user_id, + ) + + for plugin_trigger in subscribers: + workflow = workflows.get(plugin_trigger.app_id) + if not workflow: + logger.error( + "Workflow not found for app %s", + plugin_trigger.app_id, + ) + continue + + event_node = None + for node in json.loads(workflow.graph)["nodes"]: + if node["id"] == plugin_trigger.node_id and node["data"]["type"] == TRIGGER_PLUGIN_NODE_TYPE: + event_node = node["data"] + break + + if not event_node: + logger.error("Trigger event node not found for app %s", plugin_trigger.app_id) + continue + + trigger_metadata = PluginTriggerMetadata( + plugin_unique_identifier=definition.plugin_unique_identifier or "", + endpoint_id=subscription.endpoint_id, + provider_id=subscription.provider_id, + event_name=event_name, + icon_filename=definition.icon, + icon_dark_filename=definition.icon_dark, + ) + + quota_charge = unlimited() + try: + quota_charge = self._reserve_quota(QuotaType.TRIGGER, subscription.tenant_id) + except QuotaExceededError: + self._mark_rate_limited(subscription.tenant_id) + logger.info( + "Tenant %s rate limited, skipping plugin trigger %s", subscription.tenant_id, plugin_trigger.id + ) + return dispatched_count + + node_data: TriggerEventNodeData = TriggerEventNodeData.model_validate(event_node) + invoke_response: TriggerInvokeEventResponse | None = None + + try: + invoke_response = self._events.invoke(event=prepared_event, user_id=user_id, node_data=node_data) + except PluginInvokeError as e: + quota_charge.refund() + + error_message = e.to_user_friendly_error(plugin_name=definition.provider_name) + try: + end_user = end_users_by_app.get(plugin_trigger.app_id) + self._record_failure( + workflow=workflow, + plugin_trigger=plugin_trigger, + subscription=subscription, + trigger_metadata=trigger_metadata, + end_user=end_user, + error_message=error_message, + event_name=event_name, + request_id=request_id, + ) + except Exception: + logger.exception( + "Failed to record trigger failure log for app %s", + plugin_trigger.app_id, + ) + continue + except Exception: + quota_charge.refund() + + logger.exception( + "Failed to invoke trigger event for app %s", + plugin_trigger.app_id, + ) + continue + + if invoke_response is not None and invoke_response.cancelled: + quota_charge.refund() + + logger.info( + "Trigger ignored for app %s with trigger event %s", + plugin_trigger.app_id, + event_name, + ) + continue + + # Create trigger data for async execution + trigger_data = PluginTriggerData( + app_id=plugin_trigger.app_id, + tenant_id=subscription.tenant_id, + workflow_id=workflow.id, + root_node_id=plugin_trigger.node_id, + plugin_id=subscription.provider_id, + endpoint_id=subscription.endpoint_id, + inputs=invoke_response.variables, + trigger_metadata=trigger_metadata, + ) + + # Trigger async workflow + try: + end_user = end_users_by_app.get(plugin_trigger.app_id) + if not end_user: + raise ValueError(f"End user not found for app {plugin_trigger.app_id}") + + self._dispatch.trigger_workflow_async( + actor=TriggerActor(id=end_user.id, role=CreatorUserRole.END_USER), trigger_data=trigger_data + ) + quota_charge.commit() + dispatched_count += 1 + logger.info( + "Triggered workflow for app %s with trigger event %s", + plugin_trigger.app_id, + event_name, + ) + except Exception: + quota_charge.refund() + + logger.exception( + "Failed to trigger workflow for app %s", + plugin_trigger.app_id, + ) + + return dispatched_count + + def dispatch( + self, + *, + user_id: str, + events: list[str], + subscription: TriggerSubscriptionRuntime, + request_id: str, + ) -> int: + dispatched_count = 0 + for event_name in events: + try: + dispatched_count += self.dispatch_event( + user_id=user_id, + subscription=subscription, + event_name=event_name, + request_id=request_id, + ) + except Exception: + logger.exception( + "Failed to dispatch trigger '%s' for subscription %s and provider %s. Continuing...", + event_name, + subscription.id, + subscription.provider_id, + ) + # Continue processing other triggers even if one fails + continue + + logger.info( + "Completed async trigger dispatching: processed %d/%d triggers for subscription %s and provider %s", + dispatched_count, + len(events), + subscription.id, + subscription.provider_id, + ) + return dispatched_count + + def _record_failure( + self, + *, + workflow: TriggerWorkflowSnapshot, + plugin_trigger: TriggerSubscriber, + subscription: TriggerSubscriptionRuntime, + trigger_metadata: PluginTriggerMetadata, + end_user: TriggerActor | None, + error_message: str, + event_name: str, + request_id: str, + ) -> None: + actor = TriggerActor( + id=end_user.id if end_user else subscription.user_id, + role=CreatorUserRole.END_USER if end_user else CreatorUserRole.ACCOUNT, + ) + trigger_data = PluginTriggerData( + app_id=plugin_trigger.app_id, + tenant_id=subscription.tenant_id, + workflow_id=workflow.id, + root_node_id=plugin_trigger.node_id, + inputs={}, + trigger_metadata=trigger_metadata, + plugin_id=subscription.provider_id, + endpoint_id=subscription.endpoint_id, + ) + # Plan selection can call Billing; resolve it before the atomic three-log write. + queue = self._queues.get_queue(subscription.tenant_id) + self._logs.record_invocation_failure( + workflow=workflow, + actor=actor, + trigger_data=trigger_data, + error=error_message, + queue=queue, + failure_inputs={ + "event_name": event_name, + "subscription_id": subscription.id, + "request_id": request_id, + "plugin_trigger_id": plugin_trigger.id, + }, + ) diff --git a/api/services/workflow/entities.py b/api/services/workflow/entities.py index 2af0d1fd90e780..8d5a18b96247bc 100644 --- a/api/services/workflow/entities.py +++ b/api/services/workflow/entities.py @@ -3,12 +3,13 @@ """ from collections.abc import Mapping, Sequence +from dataclasses import dataclass from enum import StrEnum from typing import Any from pydantic import BaseModel, ConfigDict, Field -from models.enums import AppTriggerType, WorkflowRunTriggeredFrom +from models.enums import AppTriggerType, CreatorUserRole, WorkflowRunTriggeredFrom class AsyncTriggerStatus(StrEnum): @@ -169,3 +170,23 @@ class Strategy(StrEnum): schedule_strategy: Strategy granularity: int = Field(default=-1) # -1 means infinite + + +@dataclass(frozen=True, slots=True) +class TriggerActor: + """Identity resolved by a trigger's tenant-scoped admission/provisioning path.""" + + id: str + role: CreatorUserRole + + +@dataclass(frozen=True, slots=True) +class TriggerWorkflowSnapshot: + """Published workflow values needed after the read transaction has closed.""" + + id: str + tenant_id: str + app_id: str + type: str + version: str + graph: str diff --git a/api/services/workflow/queue_dispatcher.py b/api/services/workflow/queue_dispatcher.py index a5ffbadcf6ed1a..6a21c522bf5694 100644 --- a/api/services/workflow/queue_dispatcher.py +++ b/api/services/workflow/queue_dispatcher.py @@ -109,3 +109,27 @@ def get_dispatcher(cls, tenant_id: str) -> BaseQueueDispatcher: ) return dispatcher_class() # type: ignore + + +class CeleryWorkflowQueuePublisher: + """Resolve the existing plan queue and publish through its existing Celery task.""" + + def get_queue(self, tenant_id: str) -> str: + return QueueDispatcherManager.get_dispatcher(tenant_id).get_queue_name() + + def publish(self, *, queue: str, trigger_log_id: str) -> str: + from services.workflow.entities import WorkflowTaskData + from tasks.async_workflow_tasks import ( + execute_workflow_professional, + execute_workflow_sandbox, + execute_workflow_team, + ) + + payload = WorkflowTaskData(workflow_trigger_log_id=trigger_log_id).model_dump(mode="json") + if queue == QueuePriority.PROFESSIONAL: + task = execute_workflow_professional.delay(payload) + elif queue == QueuePriority.TEAM: + task = execute_workflow_team.delay(payload) + else: + task = execute_workflow_sandbox.delay(payload) + return task.id diff --git a/api/tasks/trigger_processing_tasks.py b/api/tasks/trigger_processing_tasks.py index 7acfad87e7e421..5c715d60c98faa 100644 --- a/api/tasks/trigger_processing_tasks.py +++ b/api/tasks/trigger_processing_tasks.py @@ -5,45 +5,17 @@ to avoid blocking the main request thread. """ -import json import logging -from collections.abc import Mapping, Sequence -from datetime import UTC, datetime -from typing import Any, Protocol +from collections.abc import Mapping +from typing import Any from celery import shared_task -from sqlalchemy import select -from sqlalchemy.orm import Session -from core.db.session_factory import session_factory -from core.plugin.entities.request import TriggerInvokeEventResponse -from core.plugin.impl.exc import PluginInvokeError -from core.trigger.constants import TRIGGER_PLUGIN_NODE_TYPE from core.trigger.debug.event_bus import TriggerDebugEventBus from core.trigger.debug.events import PluginTriggerDebugEvent, build_plugin_pool_key -from core.workflow.nodes.trigger_plugin.entities import TriggerEventNodeData -from enums import QuotaType from extensions.ext_application_services import application_services -from graphon.enums import WorkflowExecutionStatus -from models.enums import ( - AppTriggerType, - CreatorUserRole, - EndUserType, - WorkflowRunTriggeredFrom, - WorkflowTriggerStatus, -) -from models.model import App, EndUser -from models.trigger import WorkflowTriggerLog -from models.workflow import Workflow, WorkflowAppLog, WorkflowAppLogCreatedFrom, WorkflowRun -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.event_service import TriggerEventService -from services.trigger.subscription_binding_service import TriggerSubscriber, TriggerSubscriptionBindingQuery from services.trigger.subscription_query_service import TriggerSubscriptionRuntime -from services.workflow.entities import PluginTriggerData, PluginTriggerDispatchData, PluginTriggerMetadata -from services.workflow.queue_dispatcher import QueueDispatcherManager +from services.workflow.entities import PluginTriggerDispatchData logger = logging.getLogger(__name__) @@ -51,16 +23,6 @@ TRIGGER_QUEUE = "triggered_workflow_dispatcher" -class TriggerEndUserProvisioner(Protocol): - def create_end_user_batch( - self, - type: EndUserType, - tenant_id: str, - app_ids: list[str], - user_id: str, - ) -> Mapping[str, EndUser]: ... - - def dispatch_trigger_debug_event( events: list[str], user_id: str, @@ -104,353 +66,6 @@ def dispatch_trigger_debug_event( return 0 -def _get_published_workflows_by_app_ids( - session: Session, subscribers: Sequence[TriggerSubscriber] -) -> Mapping[str, Workflow]: - """Get current published workflows through apps.workflow_id.""" - app_ids = {trigger.app_id for trigger in subscribers} - tenant_ids = {trigger.tenant_id for trigger in subscribers} - if not app_ids or not tenant_ids: - return {} - - workflows = session.scalars( - select(Workflow) - .join(App, App.workflow_id == Workflow.id) - .where( - App.id.in_(app_ids), - App.tenant_id.in_(tenant_ids), - App.workflow_id.isnot(None), - Workflow.app_id == App.id, - Workflow.tenant_id == App.tenant_id, - Workflow.version != Workflow.VERSION_DRAFT, - ) - ).all() - return {w.app_id: w for w in workflows} - - -def _record_trigger_failure_log( - *, - session: Session, - workflow: Workflow, - plugin_trigger: TriggerSubscriber, - subscription: TriggerSubscriptionRuntime, - trigger_metadata: PluginTriggerMetadata, - end_user: EndUser | None, - error_message: str, - event_name: str, - request_id: str, -) -> None: - """ - Persist a workflow run, workflow app log, and trigger log entry for failed trigger invocations. - """ - now = datetime.now(UTC) - if end_user: - created_by_role = CreatorUserRole.END_USER - created_by = end_user.id - else: - created_by_role = CreatorUserRole.ACCOUNT - created_by = subscription.user_id - - failure_inputs = { - "event_name": event_name, - "subscription_id": subscription.id, - "request_id": request_id, - "plugin_trigger_id": plugin_trigger.id, - } - - workflow_run = WorkflowRun( - tenant_id=workflow.tenant_id, - app_id=workflow.app_id, - workflow_id=workflow.id, - type=workflow.type, - triggered_from=WorkflowRunTriggeredFrom.PLUGIN.value, - version=workflow.version, - graph=workflow.graph, - inputs=json.dumps(failure_inputs), - status=WorkflowExecutionStatus.FAILED.value, - outputs="{}", - error=error_message, - elapsed_time=0.0, - total_tokens=0, - total_steps=0, - created_by_role=created_by_role, - created_by=created_by, - created_at=now, - finished_at=now, - exceptions_count=0, - ) - session.add(workflow_run) - session.flush() - - workflow_app_log = WorkflowAppLog( - tenant_id=workflow.tenant_id, - app_id=workflow.app_id, - workflow_id=workflow.id, - workflow_run_id=workflow_run.id, - created_from=WorkflowAppLogCreatedFrom.SERVICE_API, - created_by_role=created_by_role, - created_by=created_by, - ) - session.add(workflow_app_log) - - dispatcher = QueueDispatcherManager.get_dispatcher(subscription.tenant_id) - queue_name = dispatcher.get_queue_name() - - trigger_data = PluginTriggerData( - app_id=plugin_trigger.app_id, - tenant_id=subscription.tenant_id, - workflow_id=workflow.id, - root_node_id=plugin_trigger.node_id, - inputs={}, - trigger_metadata=trigger_metadata, - plugin_id=subscription.provider_id, - endpoint_id=subscription.endpoint_id, - ) - - trigger_log = WorkflowTriggerLog( - tenant_id=workflow.tenant_id, - app_id=workflow.app_id, - workflow_id=workflow.id, - workflow_run_id=workflow_run.id, - root_node_id=plugin_trigger.node_id, - trigger_metadata=trigger_metadata.model_dump_json(), - trigger_type=AppTriggerType.TRIGGER_PLUGIN, - trigger_data=trigger_data.model_dump_json(), - inputs=json.dumps({}), - status=WorkflowTriggerStatus.FAILED, - error=error_message, - queue_name=queue_name, - retry_count=0, - created_by_role=created_by_role, - created_by=created_by, - triggered_at=now, - finished_at=now, - elapsed_time=0.0, - total_tokens=0, - outputs=None, - celery_task_id=None, - ) - session.add(trigger_log) - session.commit() - - -def dispatch_triggered_workflow( - user_id: str, - subscription: TriggerSubscriptionRuntime, - event_name: str, - request_id: str, - *, - end_users: TriggerEndUserProvisioner, - events: TriggerEventService, - bindings: TriggerSubscriptionBindingQuery, - trigger_status: AppTriggerService, -) -> int: - """Process triggered workflows. - - Args: - user_id: The ID of the user who activated the trigger - subscription: The trigger subscription - event_name: The name of the trigger event that was activated - request_id: The ID of the stored request in storage system - 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 - """ - subscribers = bindings.get_subscriber_triggers( - tenant_id=subscription.tenant_id, subscription_id=subscription.id, event_name=event_name - ) - if not subscribers: - logger.warning( - "No workflows found for trigger event '%s' in subscription '%s'", - event_name, - subscription.id, - ) - return 0 - - dispatched_count = 0 - prepared_event = events.prepare(subscription=subscription, event_name=event_name, request_id=request_id) - definition = prepared_event.definition - - # Ensure expire_on_commit is set to False to remain workflows available - with session_factory.create_session() as session: - workflows: Mapping[str, Workflow] = _get_published_workflows_by_app_ids(session, subscribers) - - end_users_by_app: Mapping[str, EndUser] = end_users.create_end_user_batch( - type=EndUserType.TRIGGER, - tenant_id=subscription.tenant_id, - app_ids=[plugin_trigger.app_id for plugin_trigger in subscribers], - user_id=user_id, - ) - - for plugin_trigger in subscribers: - workflow: Workflow | None = workflows.get(plugin_trigger.app_id) - if not workflow: - logger.error( - "Workflow not found for app %s", - plugin_trigger.app_id, - ) - continue - - event_node = None - for node_id, node_config in workflow.walk_nodes(TRIGGER_PLUGIN_NODE_TYPE): - if node_id == plugin_trigger.node_id: - event_node = node_config - break - - if not event_node: - logger.error("Trigger event node not found for app %s", plugin_trigger.app_id) - continue - - trigger_metadata = PluginTriggerMetadata( - plugin_unique_identifier=definition.plugin_unique_identifier or "", - endpoint_id=subscription.endpoint_id, - provider_id=subscription.provider_id, - event_name=event_name, - icon_filename=definition.icon, - icon_dark_filename=definition.icon_dark, - ) - - quota_charge = unlimited() - try: - quota_charge = QuotaService.reserve(QuotaType.TRIGGER, subscription.tenant_id) - except QuotaExceededError: - 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 - - node_data: TriggerEventNodeData = TriggerEventNodeData.model_validate(event_node) - invoke_response: TriggerInvokeEventResponse | None = None - - with session_factory.create_session() as session: - try: - invoke_response = events.invoke(event=prepared_event, user_id=user_id, node_data=node_data) - except PluginInvokeError as e: - quota_charge.refund() - - error_message = e.to_user_friendly_error(plugin_name=definition.provider_name) - try: - end_user = end_users_by_app.get(plugin_trigger.app_id) - _record_trigger_failure_log( - session=session, - workflow=workflow, - plugin_trigger=plugin_trigger, - subscription=subscription, - trigger_metadata=trigger_metadata, - end_user=end_user, - error_message=error_message, - event_name=event_name, - request_id=request_id, - ) - except Exception: - logger.exception( - "Failed to record trigger failure log for app %s", - plugin_trigger.app_id, - ) - continue - except Exception: - quota_charge.refund() - - logger.exception( - "Failed to invoke trigger event for app %s", - plugin_trigger.app_id, - ) - continue - - if invoke_response is not None and invoke_response.cancelled: - quota_charge.refund() - - logger.info( - "Trigger ignored for app %s with trigger event %s", - plugin_trigger.app_id, - event_name, - ) - continue - - # Create trigger data for async execution - trigger_data = PluginTriggerData( - app_id=plugin_trigger.app_id, - tenant_id=subscription.tenant_id, - workflow_id=workflow.id, - root_node_id=plugin_trigger.node_id, - plugin_id=subscription.provider_id, - endpoint_id=subscription.endpoint_id, - inputs=invoke_response.variables, - trigger_metadata=trigger_metadata, - ) - - # Trigger async workflow - try: - end_user = end_users_by_app.get(plugin_trigger.app_id) - if not end_user: - raise ValueError(f"End user not found for app {plugin_trigger.app_id}") - - AsyncWorkflowService.trigger_workflow_async(session=session, user=end_user, trigger_data=trigger_data) - quota_charge.commit() - dispatched_count += 1 - logger.info( - "Triggered workflow for app %s with trigger event %s", - plugin_trigger.app_id, - event_name, - ) - except Exception: - quota_charge.refund() - - logger.exception( - "Failed to trigger workflow for app %s", - plugin_trigger.app_id, - ) - - return dispatched_count - - -def dispatch_triggered_workflows( - user_id: str, - events: list[str], - subscription: TriggerSubscriptionRuntime, - request_id: str, - *, - end_users: TriggerEndUserProvisioner, - event_service: TriggerEventService, - bindings: TriggerSubscriptionBindingQuery, - trigger_status: AppTriggerService, -) -> int: - dispatched_count = 0 - for event_name in events: - try: - dispatched_count += dispatch_triggered_workflow( - user_id=user_id, - subscription=subscription, - event_name=event_name, - request_id=request_id, - end_users=end_users, - events=event_service, - bindings=bindings, - trigger_status=trigger_status, - ) - except Exception: - logger.exception( - "Failed to dispatch trigger '%s' for subscription %s and provider %s. Continuing...", - event_name, - subscription.id, - subscription.provider_id, - ) - # Continue processing other triggers even if one fails - continue - - logger.info( - "Completed async trigger dispatching: processed %d/%d triggers for subscription %s and provider %s", - dispatched_count, - len(events), - subscription.id, - subscription.provider_id, - ) - return dispatched_count - - @shared_task(queue=TRIGGER_QUEUE) def dispatch_triggered_workflows_async( dispatch_data: Mapping[str, Any], @@ -495,15 +110,11 @@ def dispatch_triggered_workflows_async( logger.error("Subscription not found: %s", subscription_id) return {"status": "failed", "error": "Subscription not found"} - workflow_dispatched = dispatch_triggered_workflows( + workflow_dispatched = application_services().workspaces.trigger_workflow_dispatch.dispatch( user_id=user_id, events=events, subscription=subscription, request_id=request_id, - 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 bc8192a6eea035..3cad548c957bec 100644 --- a/api/tasks/workflow_schedule_tasks.py +++ b/api/tasks/workflow_schedule_tasks.py @@ -10,12 +10,12 @@ ) from enums import QuotaType from extensions.ext_application_services import application_services +from models.enums import CreatorUserRole 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.schedule_service import ScheduleService -from services.workflow.entities import ScheduleTriggerData +from services.workflow.entities import ScheduleTriggerData, TriggerActor logger = logging.getLogger(__name__) @@ -52,17 +52,15 @@ def run_schedule_trigger(schedule_id: str) -> None: return try: - with session_factory.create_session() as session: - response = AsyncWorkflowService.trigger_workflow_async( - session=session, - user=tenant_owner, - trigger_data=ScheduleTriggerData( - app_id=schedule.app_id, - root_node_id=schedule.node_id, - inputs={}, - tenant_id=schedule.tenant_id, - ), - ) + response = application_services().workspaces.workflow_dispatch.trigger_workflow_async( + actor=TriggerActor(id=tenant_owner.id, role=CreatorUserRole.ACCOUNT), + trigger_data=ScheduleTriggerData( + app_id=schedule.app_id, + root_node_id=schedule.node_id, + inputs={}, + tenant_id=schedule.tenant_id, + ), + ) quota_charge.commit() logger.info("Schedule %s triggered workflow: %s", schedule_id, response.workflow_trigger_log_id) except Exception as e: 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 c66cc5de118d59..71d69b39434fc6 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 @@ -367,7 +367,7 @@ def test_trigger_workflow_execution_triggers_async_workflow_successfully( "services.trigger.webhook_service.QuotaService.reserve", return_value=quota_charge, ) as mock_reserve, - patch("services.trigger.webhook_service.AsyncWorkflowService.trigger_workflow_async") as mock_trigger, + patch("services.async_workflow_service.AsyncWorkflowService.trigger_workflow_async") as mock_trigger, ): WebhookService.trigger_workflow_execution( webhook_trigger, @@ -375,6 +375,7 @@ def test_trigger_workflow_execution_triggers_async_workflow_successfully( workflow, end_users=_EndUserServiceStub(end_user), trigger_status=application_services().apps.trigger_status, + workflow_dispatch=application_services().workspaces.workflow_dispatch, ) mock_reserve.assert_called_once() @@ -383,11 +384,10 @@ def test_trigger_workflow_execution_triggers_async_workflow_successfully( assert reserve_args[1] == webhook_trigger.tenant_id quota_charge.commit.assert_called_once() mock_trigger.assert_called_once() - trigger_args = mock_trigger.call_args.args - assert trigger_args[0] is end_user - assert trigger_args[1].workflow_id == workflow.id - assert trigger_args[1].root_node_id == webhook_trigger.node_id - assert mock_trigger.call_args.kwargs["session"] is not None + trigger_args = mock_trigger.call_args.kwargs + assert trigger_args["actor"].id == end_user.id + assert trigger_args["trigger_data"].workflow_id == workflow.id + assert trigger_args["trigger_data"].root_node_id == webhook_trigger.node_id def test_trigger_workflow_execution_marks_tenant_rate_limited_when_quota_exceeded( self, db_session_with_containers: Session, flask_app_with_containers: Flask @@ -417,6 +417,7 @@ def test_trigger_workflow_execution_marks_tenant_rate_limited_when_quota_exceede workflow, end_users=_EndUserServiceStub(_end_user(tenant_id=tenant.id, app_id=app.id)), trigger_status=application_services().apps.trigger_status, + workflow_dispatch=application_services().workspaces.workflow_dispatch, ) db_session_with_containers.refresh(app_trigger) @@ -447,6 +448,7 @@ def test_trigger_workflow_execution_logs_and_reraises_unexpected_errors( workflow, end_users=_EndUserServiceStub(RuntimeError("boom")), trigger_status=application_services().apps.trigger_status, + workflow_dispatch=application_services().workspaces.workflow_dispatch, ) assert caplog.messages.count(f"Failed to trigger workflow for webhook {webhook_trigger.webhook_id}") == 1 diff --git a/api/tests/test_containers_integration_tests/trigger/test_trigger_e2e.py b/api/tests/test_containers_integration_tests/trigger/test_trigger_e2e.py index 318c5208497610..9a10a742764da8 100644 --- a/api/tests/test_containers_integration_tests/trigger/test_trigger_e2e.py +++ b/api/tests/test_containers_integration_tests/trigger/test_trigger_e2e.py @@ -39,10 +39,11 @@ from models.workflow import Workflow from schedule import workflow_schedule_task from schedule.workflow_schedule_task import poll_workflow_schedules +from services.async_workflow_service import AsyncWorkflowService from services.system_feature_service import SystemFeatureService -from services.trigger import webhook_service from services.trigger.schedule_service import ScheduleService from services.trigger.subscription_query_service import TriggerSubscriptionRuntime +from services.workflow.entities import TriggerActor, TriggerData from services.workflow_service import WorkflowService from tasks import trigger_processing_tasks @@ -196,7 +197,10 @@ def test_webhook_trigger_creates_trigger_log( db_session_with_containers.add_all([webhook_trigger, app_trigger]) db_session_with_containers.commit() - def _fake_trigger_workflow_async(user: Any, trigger_data: Any, *, session: Session) -> SimpleNamespace: + def _fake_trigger_workflow_async( + self: AsyncWorkflowService, *, actor: TriggerActor, trigger_data: TriggerData + ) -> SimpleNamespace: + session = db_session_with_containers log = WorkflowTriggerLog( tenant_id=trigger_data.tenant_id, app_id=trigger_data.app_id, @@ -220,7 +224,7 @@ def _fake_trigger_workflow_async(user: Any, trigger_data: Any, *, session: Sessi return SimpleNamespace(workflow_trigger_log_id=log.id, task_id=None, status="queued", queue="test") monkeypatch.setattr( - webhook_service.AsyncWorkflowService, + AsyncWorkflowService, "trigger_workflow_async", _fake_trigger_workflow_async, ) @@ -515,7 +519,10 @@ def test_schedule_trigger_creates_trigger_log( db_session_with_containers.commit() # Mock AsyncWorkflowService to create WorkflowTriggerLog - def _fake_trigger_workflow_async(user: Any, trigger_data: Any, *, session: Session) -> SimpleNamespace: + def _fake_trigger_workflow_async( + self: AsyncWorkflowService, *, actor: TriggerActor, trigger_data: TriggerData + ) -> SimpleNamespace: + session = db_session_with_containers log = WorkflowTriggerLog( tenant_id=trigger_data.tenant_id, app_id=trigger_data.app_id, @@ -539,7 +546,7 @@ def _fake_trigger_workflow_async(user: Any, trigger_data: Any, *, session: Sessi return SimpleNamespace(workflow_trigger_log_id=log.id, task_id=None, status="queued", queue="test") monkeypatch.setattr( - workflow_schedule_tasks.AsyncWorkflowService, + AsyncWorkflowService, "trigger_workflow_async", _fake_trigger_workflow_async, ) diff --git a/api/tests/unit_tests/controllers/inner_api/workspace/test_workspace.py b/api/tests/unit_tests/controllers/inner_api/workspace/test_workspace.py index d667f382ad413c..dd376121293811 100644 --- a/api/tests/unit_tests/controllers/inner_api/workspace/test_workspace.py +++ b/api/tests/unit_tests/controllers/inner_api/workspace/test_workspace.py @@ -23,7 +23,11 @@ from extensions.application_services.workspace import WorkspaceServices from extensions.ext_redis import RedisClientWrapper from libs.external_api import register_external_error_handlers +from repositories.app_definition_query_repository import AppDefinitionQueryRepository +from repositories.app_scoped_end_user_repository import AppScopedEndUserRepo from repositories.credentials.query_repository import CredentialQueryRepository +from repositories.sqlalchemy_workflow_trigger_log_repository import SQLAlchemyWorkflowTriggerDispatchRepository +from repositories.trigger.app_repository import SQLAlchemyAppTriggerRepository from repositories.trigger.oauth_client_repository import TriggerOAuthClientRepository from repositories.trigger.subscription_builder_repository import RedisTriggerSubscriptionBuilderRepository from repositories.trigger.subscription_repository import ( @@ -32,6 +36,8 @@ TriggerSubscriptionRepository, ) from services.account_errors import AccountNotFoundError +from services.app_scoped_end_user_service import AppScopedEndUserService +from services.async_workflow_service import AsyncWorkflowService from services.errors.workspace import ( InvalidWorkspaceMemberRoleError, WorkspaceNotFoundError, @@ -40,6 +46,8 @@ ) from services.plugin.plugin_parameter_gateway import DaemonPluginParameterGateway from services.plugin.plugin_parameter_service import PluginParameterService +from services.quota_service import QuotaService +from services.trigger.app_trigger_service import AppTriggerService from services.trigger.endpoint_gateway import PluginTriggerEndpointGateway, enqueue_triggered_workflows from services.trigger.endpoint_service import TriggerEndpointService from services.trigger.event_gateway import PluginTriggerEventGateway @@ -64,6 +72,9 @@ from services.trigger.subscription_service import TriggerSubscriptionService from services.trigger.trigger_request_service import TriggerHttpRequestCachingService from services.trigger.trigger_subscription_builder_service import TriggerSubscriptionBuilderService +from services.trigger.workflow_dispatch_gateway import AppScopedTriggerEndUsers +from services.trigger.workflow_dispatch_service import PluginTriggerWorkflowDispatchService +from services.workflow.queue_dispatcher import CeleryWorkflowQueuePublisher from services.workspace.contracts import CreatedWorkspace, WorkspaceMembership from services.workspace.provisioning_service import WorkspaceProvisioningService from services.workspace.service import WorkspaceQueryService, WorkspaceService @@ -109,6 +120,17 @@ def provisioning(monkeypatch: pytest.MonkeyPatch, sqlite_session_factory: sessio invalidate_cache=delete_cache_for_subscription, invalidate_properties=invalidate_subscription_properties, ) + workflow_queries = AppDefinitionQueryRepository(session_factory=sqlite_session_factory) + logs = SQLAlchemyWorkflowTriggerDispatchRepository(session_factory=sqlite_session_factory) + queues = CeleryWorkflowQueuePublisher() + events = TriggerEventService( + subscriptions=subscription_queries, + plugins=PluginTriggerEventGateway(), + requests=TriggerHttpRequestCachingService, + ) + workflow_dispatch = AsyncWorkflowService( + workflows=workflow_queries, logs=logs, queues=queues, reserve_quota=QuotaService.reserve + ) services = WorkspaceServices( queries=create_autospec(WorkspaceQueryService, instance=True), management=create_autospec(WorkspaceService, instance=True), @@ -139,6 +161,24 @@ def provisioning(monkeypatch: pytest.MonkeyPatch, sqlite_session_factory: sessio plugins=PluginTriggerEventGateway(), requests=TriggerHttpRequestCachingService, ), + workflow_dispatch=workflow_dispatch, + trigger_workflow_dispatch=PluginTriggerWorkflowDispatchService( + workflows=workflow_queries, + logs=logs, + queues=queues, + dispatch=workflow_dispatch, + events=events, + bindings=TriggerSubscriptionRepository(session_factory=sqlite_session_factory), + end_users=AppScopedTriggerEndUsers( + end_users=AppScopedEndUserService( + end_users=AppScopedEndUserRepo(session_factory=sqlite_session_factory) + ) + ), + reserve_quota=QuotaService.reserve, + mark_rate_limited=AppTriggerService( + triggers=SQLAlchemyAppTriggerRepository(session_factory=sqlite_session_factory) + ).mark_tenant_triggers_rate_limited, + ), trigger_builders=builders, trigger_endpoints=TriggerEndpointService( subscriptions=TriggerSubscriptionRepository(session_factory=sqlite_session_factory), diff --git a/api/tests/unit_tests/controllers/trigger/test_webhook.py b/api/tests/unit_tests/controllers/trigger/test_webhook.py index 760e94e3b1dbf5..a6c847d6f5161a 100644 --- a/api/tests/unit_tests/controllers/trigger/test_webhook.py +++ b/api/tests/unit_tests/controllers/trigger/test_webhook.py @@ -30,10 +30,16 @@ class _AppServicesStub: trigger_status: AppTriggerService +@dataclass(eq=False, repr=False) +class _WorkspaceServicesStub: + workflow_dispatch: object + + class _ApplicationServicesStub: def __init__(self, commands: object, trigger_status: AppTriggerService) -> None: self.app_scoped_end_users = _AppScopedEndUserServicesStub(commands) self.apps = _AppServicesStub(trigger_status) + self.workspaces = _WorkspaceServicesStub(object()) @pytest.fixture(autouse=True) diff --git a/api/tests/unit_tests/repositories/test_trigger_execution_logs.py b/api/tests/unit_tests/repositories/test_trigger_execution_logs.py new file mode 100644 index 00000000000000..2c6c4fd8ebc3db --- /dev/null +++ b/api/tests/unit_tests/repositories/test_trigger_execution_logs.py @@ -0,0 +1,83 @@ +"""Producer log transactions are scoped and preserve all-or-nothing failure records.""" + +from uuid import uuid4 + +import pytest +from sqlalchemy import Table, select +from sqlalchemy.exc import OperationalError + +from models.enums import CreatorUserRole, WorkflowTriggerStatus +from models.trigger import WorkflowTriggerLog +from models.workflow import WorkflowAppLog, WorkflowRun +from repositories.app_definition_query_repository import AppDefinitionQueryRepository +from repositories.sqlalchemy_workflow_trigger_log_repository import SQLAlchemyWorkflowTriggerDispatchRepository +from services.workflow.entities import ScheduleTriggerData, TriggerActor +from tests.unit_tests.controllers.console.workspace.test_trigger_oauth_client import _Harness +from tests.unit_tests.tasks import test_trigger_processing_tasks as worker_tests +from tests.unit_tests.tasks.test_trigger_processing_tasks import _Dispatch + +application_redis = worker_tests.application_redis +harness = worker_tests.harness +dispatch = worker_tests.dispatch + + +@pytest.mark.parametrize("scope", ["tenant", "app"]) +def test_status_write_does_not_update_log_outside_owner_scope( + harness: _Harness, + dispatch: _Dispatch, + scope: str, +) -> None: + store = SQLAlchemyWorkflowTriggerDispatchRepository(session_factory=harness.session_factory) + data = ScheduleTriggerData(tenant_id=harness.tenant_id, app_id=dispatch.app_id, root_node_id="node", inputs={}) + workflow = AppDefinitionQueryRepository(session_factory=harness.session_factory).get_trigger_workflow( + tenant_id=harness.tenant_id, app_id=dispatch.app_id, workflow_id=None + ) + log_id = store.create_pending( + actor=TriggerActor(id=harness.account_id, role=CreatorUserRole.ACCOUNT), + trigger_data=data, + workflow_id=workflow.id, + queue="workflow_sandbox", + ) + tenant_id = str(uuid4()) if scope == "tenant" else harness.tenant_id + app_id = str(uuid4()) if scope == "app" else dispatch.app_id + store.mark_failed( + tenant_id=tenant_id, app_id=app_id, log_id=log_id, status=WorkflowTriggerStatus.FAILED, error="foreign" + ) + store.mark_queued(tenant_id=tenant_id, app_id=app_id, log_id=log_id, task_id="foreign") + with harness.session_factory() as session: + log = session.get(WorkflowTriggerLog, log_id) + assert log is not None + assert log.status == WorkflowTriggerStatus.PENDING + assert log.error is None + assert log.celery_task_id is None + + +def test_three_failure_records_rollback_together_when_app_log_write_fails( + harness: _Harness, + dispatch: _Dispatch, +) -> None: + store = SQLAlchemyWorkflowTriggerDispatchRepository(session_factory=harness.session_factory) + workflow = AppDefinitionQueryRepository(session_factory=harness.session_factory).get_trigger_workflow( + tenant_id=harness.tenant_id, app_id=dispatch.app_id, workflow_id=None + ) + data = ScheduleTriggerData(tenant_id=harness.tenant_id, app_id=dispatch.app_id, root_node_id="node", inputs={}) + with harness.session_factory() as session: + engine = session.get_bind() + app_logs = WorkflowAppLog.__table__ + assert isinstance(app_logs, Table) + app_logs.drop(engine) + try: + with pytest.raises(OperationalError, match="workflow_app_logs"): + store.record_invocation_failure( + workflow=workflow, + actor=TriggerActor(id=harness.account_id, role=CreatorUserRole.ACCOUNT), + trigger_data=data, + failure_inputs={"event_name": "push"}, + error="provider rejected", + queue="workflow", + ) + with harness.session_factory() as session: + assert session.scalar(select(WorkflowRun)) is None + assert session.scalar(select(WorkflowTriggerLog)) is None + finally: + app_logs.create(engine) diff --git a/api/tests/unit_tests/services/test_async_workflow_service.py b/api/tests/unit_tests/services/test_async_workflow_service.py index 510a381152421d..401cf9b961606d 100644 --- a/api/tests/unit_tests/services/test_async_workflow_service.py +++ b/api/tests/unit_tests/services/test_async_workflow_service.py @@ -1,631 +1,203 @@ -import json -import logging -from datetime import UTC, datetime, timedelta -from types import SimpleNamespace -from unittest.mock import MagicMock, patch +"""Real producer persistence, Billing HTTP and Celery transport contracts.""" + +from contextvars import Context +from uuid import uuid4 import pytest +from celery import Celery +from kombu.exceptions import SerializerNotInstalled +from kombu.simple import SimpleQueue from sqlalchemy import select -from sqlalchemy.engine import Engine -from sqlalchemy.orm import Session -import services.async_workflow_service as async_workflow_service_module -from models.account import Account -from models.enums import AppTriggerType, CreatorUserRole, EndUserType, WorkflowRunTriggeredFrom, WorkflowTriggerStatus -from models.model import App, AppMode, EndUser +from enums import DeploymentEdition +from extensions.ext_application_services import application_services +from models.enums import CreatorUserRole, WorkflowTriggerStatus +from models.model import App from models.trigger import WorkflowTriggerLog -from models.workflow import Workflow, WorkflowType -from services.async_workflow_service import AsyncWorkflowService -from services.errors.app import QuotaExceededError, WorkflowNotFoundError -from services.workflow.entities import AsyncTriggerResponse, TriggerData -from services.workflow.queue_dispatcher import QueuePriority - - -class AsyncWorkflowServiceTestDataFactory: - """Factory helpers for async workflow service unit tests.""" - - @staticmethod - def create_trigger_data( - app_id: str = "app-123", - tenant_id: str = "tenant-123", - workflow_id: str | None = "workflow-123", - root_node_id: str = "root-node-123", - ) -> TriggerData: - """Create valid trigger data for async workflow execution tests.""" - return TriggerData( - app_id=app_id, - tenant_id=tenant_id, - workflow_id=workflow_id, - root_node_id=root_node_id, - inputs={"name": "dify"}, - files=[], - trigger_type=AppTriggerType.UNKNOWN, - trigger_from=WorkflowRunTriggeredFrom.APP_RUN, - trigger_metadata=None, - ) - - @staticmethod - def create_app(app_id: str = "app-123", tenant_id: str = "tenant-123") -> App: - """Create an app that can be persisted for trigger lookup tests.""" - return App( - id=app_id, - tenant_id=tenant_id, - name="Async workflow app", - description="", - mode=AppMode.WORKFLOW, - enable_site=True, - enable_api=True, - max_active_requests=0, - ) - - @staticmethod - def create_trigger_log_with_data( - trigger_data: TriggerData, - *, - trigger_log_id: str = "trigger-log-123", - retry_count: int = 0, - status: WorkflowTriggerStatus = WorkflowTriggerStatus.FAILED, - created_at: datetime | None = None, - ) -> WorkflowTriggerLog: - """Create a persistent trigger log with serialized trigger data.""" - trigger_log = WorkflowTriggerLog( - tenant_id=trigger_data.tenant_id, - app_id=trigger_data.app_id, - workflow_id=trigger_data.workflow_id or "workflow-123", - workflow_run_id=None, - root_node_id=trigger_data.root_node_id, - trigger_metadata="{}", - trigger_type=trigger_data.trigger_type, - trigger_data=trigger_data.model_dump_json(), - inputs=json.dumps(dict(trigger_data.inputs)), - outputs=None, - status=status, - error="previous-error", - queue_name=QueuePriority.SANDBOX, - celery_task_id=None, - created_by_role=CreatorUserRole.END_USER, - created_by="end-user-123", - retry_count=retry_count, - elapsed_time=None, - total_tokens=None, - triggered_at=None, - finished_at=None, - ) - trigger_log.id = trigger_log_id - if created_at is not None: - trigger_log.created_at = created_at - return trigger_log - - @staticmethod - def create_workflow( - *, workflow_id: str = "workflow-123", app_id: str = "app-123", tenant_id: str = "tenant-123" - ) -> Workflow: - """Create a mapped workflow for service-return and trigger-log tests.""" - return Workflow( - id=workflow_id, - tenant_id=tenant_id, - app_id=app_id, - type=WorkflowType.WORKFLOW, - version="1", - graph="{}", - _features="{}", - created_by="account-123", - ) - - @staticmethod - def create_account(account_id: str = "account-123") -> Account: - """Create a mapped account for role-discrimination tests.""" - account = Account(name="Account", email=f"{account_id}@example.com") - account.id = account_id - return account - - @staticmethod - def create_end_user(end_user_id: str = "end-user-123") -> EndUser: - """Create a mapped end user for role-discrimination and retry tests.""" - return EndUser( - id=end_user_id, - tenant_id="tenant-123", - app_id="app-123", - type=EndUserType.BROWSER, - name="End User", - session_id=f"session-{end_user_id}", - ) - - -@pytest.mark.usefixtures("sqlite_session") -@pytest.mark.parametrize("sqlite_session", [(App, WorkflowTriggerLog)], indirect=True) -class TestAsyncWorkflowService: - @pytest.fixture - def async_workflow_trigger_mocks(self): - """Shared fixture for async workflow trigger tests. - - Yields mocks for: - - dispatcher_manager_class: QueueDispatcherManager class - - dispatcher: dispatcher instance - - quota_service: QuotaService mock - - get_workflow: AsyncWorkflowService._get_workflow method - - professional_task: execute_workflow_professional - - team_task: execute_workflow_team - - sandbox_task: execute_workflow_sandbox - """ - mock_dispatcher = MagicMock() - mock_quota_service = MagicMock() - - with ( - patch.object(async_workflow_service_module, "QueueDispatcherManager") as mock_dispatcher_manager_class, - patch.object(async_workflow_service_module, "WorkflowService"), - patch.object( - async_workflow_service_module.AsyncWorkflowService, - "_get_workflow", - ) as mock_get_workflow, - patch.object( - async_workflow_service_module, - "QuotaService", - new=mock_quota_service, - ), - patch.object(async_workflow_service_module, "execute_workflow_professional") as mock_professional_task, - patch.object(async_workflow_service_module, "execute_workflow_team") as mock_team_task, - patch.object(async_workflow_service_module, "execute_workflow_sandbox") as mock_sandbox_task, - ): - # Configure dispatcher_manager to return our mock_dispatcher - mock_dispatcher_manager_class.return_value.get_dispatcher.return_value = mock_dispatcher - - yield { - "dispatcher_manager_class": mock_dispatcher_manager_class, - "dispatcher": mock_dispatcher, - "quota_service": mock_quota_service, - "get_workflow": mock_get_workflow, - "professional_task": mock_professional_task, - "team_task": mock_team_task, - "sandbox_task": mock_sandbox_task, - } - - @pytest.mark.parametrize( - ("queue_name", "selected_task_attr"), - [ - (QueuePriority.PROFESSIONAL, "execute_workflow_professional"), - (QueuePriority.TEAM, "execute_workflow_team"), - (QueuePriority.SANDBOX, "execute_workflow_sandbox"), - ], - ) - def test_should_dispatch_to_matching_celery_task_when_triggering_workflow( - self, - queue_name, - selected_task_attr, - async_workflow_trigger_mocks, - sqlite_session: Session, - ): - """Test queue-based task routing and successful async trigger response.""" - # Arrange - sqlite_session.add(AsyncWorkflowServiceTestDataFactory.create_app()) - sqlite_session.commit() - trigger_data = AsyncWorkflowServiceTestDataFactory.create_trigger_data() - workflow = AsyncWorkflowServiceTestDataFactory.create_workflow() - - mocks = async_workflow_trigger_mocks - mocks["dispatcher"].get_queue_name.return_value = queue_name - mocks["get_workflow"].return_value = workflow - - task_result = MagicMock() - task_result.id = "task-123" - mocks["professional_task"].delay.return_value = task_result - mocks["team_task"].delay.return_value = task_result - mocks["sandbox_task"].delay.return_value = task_result - - quota_charge_mock = MagicMock() - mocks["quota_service"].reserve.return_value = quota_charge_mock - - user = AsyncWorkflowServiceTestDataFactory.create_account() - - # Act - result = AsyncWorkflowService.trigger_workflow_async( - session=sqlite_session, user=user, trigger_data=trigger_data - ) - - # Assert - assert isinstance(result, AsyncTriggerResponse) - assert result.workflow_trigger_log_id - assert result.task_id == "task-123" - assert result.status == "queued" - assert result.queue == queue_name - - mocks["quota_service"].reserve.assert_called_once() - quota_charge_mock.commit.assert_called_once() - assert not sqlite_session.in_transaction() - - created_log = sqlite_session.scalar( - select(WorkflowTriggerLog).where(WorkflowTriggerLog.id == result.workflow_trigger_log_id) - ) - assert created_log is not None - assert created_log.status == WorkflowTriggerStatus.QUEUED - assert created_log.queue_name == queue_name - assert created_log.created_by_role == CreatorUserRole.ACCOUNT - assert created_log.created_by == "account-123" - assert created_log.trigger_data == trigger_data.model_dump_json() - assert created_log.inputs == json.dumps(dict(trigger_data.inputs)) - assert created_log.celery_task_id == "task-123" - - task_mocks = { - "execute_workflow_professional": mocks["professional_task"], - "execute_workflow_team": mocks["team_task"], - "execute_workflow_sandbox": mocks["sandbox_task"], - } - for task_attr, task_mock in task_mocks.items(): - if task_attr == selected_task_attr: - task_mock.delay.assert_called_once_with({"workflow_trigger_log_id": result.workflow_trigger_log_id}) - else: - task_mock.delay.assert_not_called() - - def test_should_set_end_user_role_when_triggered_by_end_user( - self, async_workflow_trigger_mocks, sqlite_session: Session - ): - """Test that non-account users are tracked as END_USER in trigger logs.""" - # Arrange - sqlite_session.add(AsyncWorkflowServiceTestDataFactory.create_app()) - sqlite_session.commit() - trigger_data = AsyncWorkflowServiceTestDataFactory.create_trigger_data() - workflow = AsyncWorkflowServiceTestDataFactory.create_workflow() - - mocks = async_workflow_trigger_mocks - mocks["dispatcher"].get_queue_name.return_value = QueuePriority.SANDBOX - mocks["get_workflow"].return_value = workflow - - task_result = MagicMock(id="task-123") - mocks["sandbox_task"].delay.return_value = task_result - - user = AsyncWorkflowServiceTestDataFactory.create_end_user() - - # Act - response = AsyncWorkflowService.trigger_workflow_async( - session=sqlite_session, user=user, trigger_data=trigger_data - ) - - # Assert - created_log = sqlite_session.get(WorkflowTriggerLog, response.workflow_trigger_log_id) - assert created_log is not None - assert created_log.created_by_role == CreatorUserRole.END_USER - assert created_log.created_by == "end-user-123" - - def test_should_raise_workflow_not_found_when_app_does_not_exist(self, sqlite_session: Session): - """Test trigger failure when app lookup returns no result.""" - # Arrange - trigger_data = AsyncWorkflowServiceTestDataFactory.create_trigger_data(app_id="missing-app") - - with ( - patch.object(async_workflow_service_module, "QueueDispatcherManager"), - patch.object(async_workflow_service_module, "WorkflowService"), - ): - # Act / Assert - with pytest.raises(WorkflowNotFoundError, match="App not found: missing-app"): - AsyncWorkflowService.trigger_workflow_async( - session=sqlite_session, - user=AsyncWorkflowServiceTestDataFactory.create_end_user("user-123"), - trigger_data=trigger_data, - ) - - def test_should_mark_log_rate_limited_and_reraise_when_quota_exceeded( - self, - async_workflow_trigger_mocks, - caplog: pytest.LogCaptureFixture, - sqlite_session: Session, - ): - """Test quota-exceeded path updates trigger log and preserves the quota exception.""" - # Arrange - sqlite_session.add(AsyncWorkflowServiceTestDataFactory.create_app()) - sqlite_session.commit() - trigger_data = AsyncWorkflowServiceTestDataFactory.create_trigger_data() - workflow = AsyncWorkflowServiceTestDataFactory.create_workflow() - - mocks = async_workflow_trigger_mocks - mocks["dispatcher"].get_queue_name.return_value = QueuePriority.TEAM - mocks["get_workflow"].return_value = workflow - mocks["quota_service"].reserve.side_effect = QuotaExceededError( - feature="workflow", - tenant_id="tenant-123", - required=1, - ) - caplog.set_level(logging.INFO, logger=async_workflow_service_module.__name__) - - # Act / Assert - with pytest.raises(QuotaExceededError) as exc_info: - AsyncWorkflowService.trigger_workflow_async( - session=sqlite_session, - user=AsyncWorkflowServiceTestDataFactory.create_end_user("user-123"), - trigger_data=trigger_data, - ) - - assert exc_info.value.feature == "workflow" - assert exc_info.value.tenant_id == "tenant-123" - assert exc_info.value.required == 1 - assert not sqlite_session.in_transaction() - updated_log = sqlite_session.scalar(select(WorkflowTriggerLog)) - assert updated_log is not None - assert updated_log.status == WorkflowTriggerStatus.RATE_LIMITED - assert "Quota limit reached" in updated_log.error - assert ( - "Workflow quota exceeded for tenant tenant-123, app app-123, workflow workflow-123, " - f"trigger log {updated_log.id}" - ) in caplog.messages - mocks["professional_task"].delay.assert_not_called() - mocks["team_task"].delay.assert_not_called() - mocks["sandbox_task"].delay.assert_not_called() - - def test_should_mark_log_failed_and_reraise_when_celery_dispatch_raises( - self, - async_workflow_trigger_mocks, - caplog: pytest.LogCaptureFixture, - sqlite_session: Session, - ): - """Test dispatch failure path marks trigger log FAILED, refunds quota, and re-raises.""" - # Arrange - sqlite_session.add(AsyncWorkflowServiceTestDataFactory.create_app()) - sqlite_session.commit() - trigger_data = AsyncWorkflowServiceTestDataFactory.create_trigger_data() - workflow = AsyncWorkflowServiceTestDataFactory.create_workflow() - - mocks = async_workflow_trigger_mocks - mocks["dispatcher"].get_queue_name.return_value = QueuePriority.SANDBOX - mocks["get_workflow"].return_value = workflow - - quota_charge_mock = MagicMock() - mocks["quota_service"].reserve.return_value = quota_charge_mock - - dispatch_error = RuntimeError("broker unavailable") - mocks["sandbox_task"].delay.side_effect = dispatch_error - caplog.set_level(logging.ERROR, logger=async_workflow_service_module.__name__) - - # Act / Assert - with pytest.raises(RuntimeError) as exc_info: - AsyncWorkflowService.trigger_workflow_async( - session=sqlite_session, - user=AsyncWorkflowServiceTestDataFactory.create_end_user("user-123"), - trigger_data=trigger_data, - ) - - assert exc_info.value is dispatch_error - quota_charge_mock.commit.assert_not_called() - quota_charge_mock.refund.assert_called_once() - assert not sqlite_session.in_transaction() - - updated_log = sqlite_session.scalar(select(WorkflowTriggerLog)) - assert updated_log is not None - assert updated_log.status == WorkflowTriggerStatus.FAILED - assert updated_log.error is not None - assert "Failed to dispatch workflow task" in updated_log.error - assert "broker unavailable" in updated_log.error - assert ( - "Failed to dispatch workflow task for tenant tenant-123, app app-123, workflow workflow-123, " - f"trigger log {updated_log.id}" - ) in caplog.text - - def test_should_raise_when_reinvoke_target_log_does_not_exist(self, sqlite_session: Session): - """Test reinvoke_trigger error path when original trigger log is missing.""" - # Arrange - # Act / Assert - with pytest.raises(ValueError, match="Trigger log not found: missing-log"): - AsyncWorkflowService.reinvoke_trigger( - session=sqlite_session, - user=AsyncWorkflowServiceTestDataFactory.create_end_user("user-123"), - workflow_trigger_log_id="missing-log", +from models.workflow import Workflow +from services.errors.app import IsDraftWorkflowError, QuotaExceededError, WorkflowNotFoundError +from services.workflow.entities import AsyncTriggerResponse, ScheduleTriggerData, TriggerActor +from tasks.async_workflow_tasks import execute_workflow_professional, execute_workflow_sandbox, execute_workflow_team +from tasks.workflow_cfs_scheduler.entities import AsyncWorkflowQueue +from tests.unit_tests.config_override import config_overrides_context +from tests.unit_tests.controllers.console.workspace.test_trigger_oauth_client import _Harness +from tests.unit_tests.tasks import test_trigger_processing_tasks as worker_tests +from tests.unit_tests.tasks.test_trigger_processing_tasks import _Billing, _Dispatch + +application_redis = worker_tests.application_redis +harness = worker_tests.harness +billing = worker_tests.billing +dispatch = worker_tests.dispatch + + +def _run( + harness: _Harness, + dispatch: _Dispatch, + *, + app_id: str | None = None, + tenant_id: str | None = None, + workflow_id: str | None = None, + actor: TriggerActor | None = None, +) -> AsyncTriggerResponse: + def invoke() -> AsyncTriggerResponse: + with harness.app.app_context(), config_overrides_context(DEPLOYMENT_EDITION=DeploymentEdition.CLOUD): + return application_services().workspaces.workflow_dispatch.trigger_workflow_async( + actor=actor or TriggerActor(id=harness.account_id, role=CreatorUserRole.ACCOUNT), + trigger_data=ScheduleTriggerData( + tenant_id=tenant_id or harness.tenant_id, + app_id=app_id or dispatch.app_id, + workflow_id=workflow_id, + root_node_id="push-node", + inputs={"text": "hello"}, + ), ) - def test_should_update_original_log_and_requeue_when_reinvoking(self, sqlite_session: Session): - """Test reinvoke flow updates original log state and triggers a new async run.""" - # Arrange - trigger_data = AsyncWorkflowServiceTestDataFactory.create_trigger_data() - trigger_log = AsyncWorkflowServiceTestDataFactory.create_trigger_log_with_data(trigger_data, retry_count=1) - sqlite_session.add(trigger_log) - sqlite_session.commit() - - expected_response = AsyncTriggerResponse( - workflow_trigger_log_id="new-trigger-log-456", - task_id="task-456", - status="queued", - queue=QueuePriority.TEAM, - ) - + return Context().run(invoke) + + +@pytest.mark.parametrize("role", [CreatorUserRole.ACCOUNT, CreatorUserRole.END_USER]) +@pytest.mark.parametrize("plan", ["sandbox", "professional", "team", "unknown-plan"]) +def test_records_actor_and_publishes_lightweight_task( + harness: _Harness, + dispatch: _Dispatch, + billing: _Billing, + role: CreatorUserRole, + plan: str, +) -> None: + billing.plan = plan + expected_task = {"professional": execute_workflow_professional, "team": execute_workflow_team}.get( + plan, execute_workflow_sandbox + ) + previous = execute_workflow_sandbox.app + celery = Celery("workflow-producer-test", broker="memory://", set_as_current=False) + celery.conf.update(task_always_eager=False, task_serializer="json") + celery.set_current() + actor = TriggerActor(id=str(uuid4()), role=role) + try: with ( - patch.object( - async_workflow_service_module.AsyncWorkflowService, - "trigger_workflow_async", - return_value=expected_response, - ) as mock_trigger_workflow_async, + celery.connection_for_write() as connection, + SimpleQueue(connection.channel(), AsyncWorkflowQueue.SANDBOX_QUEUE) as queue, ): - user = AsyncWorkflowServiceTestDataFactory.create_end_user("user-123") - - # Act - response = AsyncWorkflowService.reinvoke_trigger( - session=sqlite_session, - user=user, - workflow_trigger_log_id="trigger-log-123", - ) - - # Assert - assert response == expected_response - assert not sqlite_session.in_transaction() - sqlite_session.refresh(trigger_log) - assert trigger_log.status == WorkflowTriggerStatus.RETRYING - assert trigger_log.retry_count == 2 - assert trigger_log.error is None - assert trigger_log.triggered_at is not None - called_trigger_data = mock_trigger_workflow_async.call_args.args[1] - assert isinstance(called_trigger_data, TriggerData) - assert called_trigger_data.app_id == "app-123" - - @pytest.mark.parametrize( - ("lookup_id", "tenant_id", "expected_id"), - [ - ("missing-log", "tenant-123", None), - ("trigger-log-123", "tenant-123", "trigger-log-123"), - ("trigger-log-123", "other-tenant", None), - ], - ) - def test_should_return_trigger_log_dict_or_none( - self, - lookup_id: str, - tenant_id: str, - expected_id: str | None, - sqlite_session: Session, - sqlite_engine: Engine, - ): - """Test get_trigger_log returns persisted data with tenant isolation.""" - # Arrange - trigger_data = AsyncWorkflowServiceTestDataFactory.create_trigger_data() - sqlite_session.add(AsyncWorkflowServiceTestDataFactory.create_trigger_log_with_data(trigger_data)) - sqlite_session.commit() - - with patch.object(async_workflow_service_module, "db", SimpleNamespace(engine=sqlite_engine)): - # Act - result = AsyncWorkflowService.get_trigger_log(lookup_id, tenant_id=tenant_id) - - # Assert - assert (result["id"] if result else None) == expected_id - - def test_should_return_recent_logs_as_dict_list(self, sqlite_session: Session, sqlite_engine: Engine): - """Test recent logs are ordered, paginated, and tenant/app scoped.""" - # Arrange - now = datetime.now(UTC) - logs = [ - AsyncWorkflowServiceTestDataFactory.create_trigger_log_with_data( - AsyncWorkflowServiceTestDataFactory.create_trigger_data(), - trigger_log_id=f"log-{index}", - created_at=now - timedelta(minutes=index), - ) - for index in range(1, 4) - ] - logs.extend( - [ - AsyncWorkflowServiceTestDataFactory.create_trigger_log_with_data( - AsyncWorkflowServiceTestDataFactory.create_trigger_data(tenant_id="other-tenant"), - trigger_log_id="other-tenant-log", - created_at=now, - ), - AsyncWorkflowServiceTestDataFactory.create_trigger_log_with_data( - AsyncWorkflowServiceTestDataFactory.create_trigger_data(app_id="other-app"), - trigger_log_id="other-app-log", - created_at=now, - ), - ] - ) - sqlite_session.add_all(logs) - sqlite_session.commit() - - with patch.object(async_workflow_service_module, "db", SimpleNamespace(engine=sqlite_engine)): - # Act - result = AsyncWorkflowService.get_recent_logs( - tenant_id="tenant-123", - app_id="app-123", - hours=12, - limit=2, - offset=1, - ) - - # Assert - assert [log["id"] for log in result] == ["log-2", "log-3"] - - def test_should_return_failed_logs_for_retry_as_dict_list(self, sqlite_session: Session, sqlite_engine: Engine): - """Test retry candidates are status, retry-count, and tenant scoped.""" - # Arrange - now = datetime.now(UTC) - candidates = [ - AsyncWorkflowServiceTestDataFactory.create_trigger_log_with_data( - AsyncWorkflowServiceTestDataFactory.create_trigger_data(), - trigger_log_id="failed-log-1", - retry_count=1, - created_at=now - timedelta(minutes=2), - ), - AsyncWorkflowServiceTestDataFactory.create_trigger_log_with_data( - AsyncWorkflowServiceTestDataFactory.create_trigger_data(), - trigger_log_id="rate-limited-log", - retry_count=2, - status=WorkflowTriggerStatus.RATE_LIMITED, - created_at=now - timedelta(minutes=1), - ), - AsyncWorkflowServiceTestDataFactory.create_trigger_log_with_data( - AsyncWorkflowServiceTestDataFactory.create_trigger_data(), - trigger_log_id="retry-limit-log", - retry_count=4, - ), - AsyncWorkflowServiceTestDataFactory.create_trigger_log_with_data( - AsyncWorkflowServiceTestDataFactory.create_trigger_data(tenant_id="other-tenant"), - trigger_log_id="other-tenant-log", - ), - AsyncWorkflowServiceTestDataFactory.create_trigger_log_with_data( - AsyncWorkflowServiceTestDataFactory.create_trigger_data(), - trigger_log_id="queued-log", - status=WorkflowTriggerStatus.QUEUED, - ), - ] - sqlite_session.add_all(candidates) - sqlite_session.commit() - - with patch.object(async_workflow_service_module, "db", SimpleNamespace(engine=sqlite_engine)): - # Act - result = AsyncWorkflowService.get_failed_logs_for_retry(tenant_id="tenant-123", max_retry_count=4, limit=20) - - # Assert - assert [log["id"] for log in result] == ["failed-log-1", "rate-limited-log"] - - -@pytest.mark.usefixtures("sqlite_session") -@pytest.mark.parametrize("sqlite_session", [(App, WorkflowTriggerLog)], indirect=True) -class TestAsyncWorkflowServiceGetWorkflow: - def test_should_return_specific_workflow_when_workflow_id_exists(self, sqlite_session: Session): - """Test _get_workflow returns published workflow by id when provided.""" - # Arrange - workflow_service = MagicMock() - app_model = AsyncWorkflowServiceTestDataFactory.create_app() - workflow = AsyncWorkflowServiceTestDataFactory.create_workflow() - workflow_service.get_published_workflow_by_id.return_value = workflow - - # Act - result = AsyncWorkflowService._get_workflow( - workflow_service, app_model, workflow_id="workflow-123", session=sqlite_session - ) - - # Assert - assert result == workflow - workflow_service.get_published_workflow_by_id.assert_called_once_with( - app_model, "workflow-123", session=sqlite_session - ) - workflow_service.get_published_workflow.assert_not_called() - - def test_should_raise_when_specific_workflow_id_not_found(self, sqlite_session: Session): - """Test _get_workflow raises WorkflowNotFoundError for unknown workflow id.""" - # Arrange - workflow_service = MagicMock() - app_model = AsyncWorkflowServiceTestDataFactory.create_app() - workflow_service.get_published_workflow_by_id.return_value = None - - # Act / Assert - with pytest.raises(WorkflowNotFoundError, match="Published workflow not found: workflow-404"): - AsyncWorkflowService._get_workflow( - workflow_service, app_model, workflow_id="workflow-404", session=sqlite_session - ) - - def test_should_return_default_published_workflow_when_workflow_id_not_provided(self, sqlite_session: Session): - """Test _get_workflow returns default published workflow when no id is provided.""" - # Arrange - workflow_service = MagicMock() - app_model = AsyncWorkflowServiceTestDataFactory.create_app() - workflow = AsyncWorkflowServiceTestDataFactory.create_workflow() - workflow_service.get_published_workflow.return_value = workflow - - # Act - result = AsyncWorkflowService._get_workflow(workflow_service, app_model, session=sqlite_session) - - # Assert - assert result == workflow - workflow_service.get_published_workflow.assert_called_once_with(app_model, session=sqlite_session) - workflow_service.get_published_workflow_by_id.assert_not_called() - - def test_should_raise_when_default_published_workflow_not_found(self, sqlite_session: Session): - """Test _get_workflow raises WorkflowNotFoundError when app has no published workflow.""" - # Arrange - workflow_service = MagicMock() - app_model = AsyncWorkflowServiceTestDataFactory.create_app() - workflow_service.get_published_workflow.return_value = None - - # Act / Assert - with pytest.raises(WorkflowNotFoundError, match="No published workflow found for app: app-123"): - AsyncWorkflowService._get_workflow(workflow_service, app_model, session=sqlite_session) + queue.clear() + try: + result = _run(harness, dispatch, actor=actor) + message = queue.get(block=False) + with harness.session_factory() as session: + log = session.get(WorkflowTriggerLog, result.workflow_trigger_log_id) + assert log is not None + assert log.created_by == actor.id + assert log.created_by_role == role + assert log.status == WorkflowTriggerStatus.QUEUED + assert log.celery_task_id == result.task_id + assert log.triggered_at is not None + assert message.payload[0] == [{"workflow_trigger_log_id": log.id}] + assert message.headers["id"] == log.celery_task_id + assert message.headers["task"] == expected_task.name + assert result.queue == f"workflow_{plan if plan != 'unknown-plan' else 'sandbox'}" + message.ack() + assert queue.qsize() == 0 + finally: + queue.clear() + finally: + previous.set_current() + celery.close() + assert [(path, payload["feature_key"]) for path, payload in billing.requests] == [ + ("/quota/reserve", "api_rate_limit"), + ("/quota/commit", "api_rate_limit"), + ] + assert all(not session.in_transaction() for session in harness.metadata.sessions) + assert billing.open_transactions + assert not any(billing.open_transactions) + + +def test_quota_failure_persists_rate_limited_log_without_dispatch( + harness: _Harness, + dispatch: _Dispatch, + billing: _Billing, +) -> None: + billing.exhausted = True + with pytest.raises(QuotaExceededError): + _run(harness, dispatch) + with harness.session_factory() as session: + log = session.scalar(select(WorkflowTriggerLog)) + assert log is not None + assert log.status == WorkflowTriggerStatus.RATE_LIMITED + assert log.error is not None + assert log.error.startswith("Quota limit reached:") + assert log.celery_task_id is None + assert [path for path, _ in billing.requests] == ["/quota/reserve"] + + +def test_serialization_failure_records_failed_log_and_refunds_quota( + harness: _Harness, + dispatch: _Dispatch, + billing: _Billing, +) -> None: + previous = execute_workflow_sandbox.app + celery = Celery("workflow-producer-failure", broker="memory://", set_as_current=False) + celery.conf.update(task_always_eager=False, task_serializer="unregistered-trigger-test-serializer") + celery.set_current() + try: + with pytest.raises(SerializerNotInstalled): + _run(harness, dispatch) + finally: + previous.set_current() + celery.close() + with harness.session_factory() as session: + log = session.scalar(select(WorkflowTriggerLog)) + assert log is not None + assert log.status == WorkflowTriggerStatus.FAILED + assert log.error is not None + assert log.error.startswith("Failed to dispatch workflow task:") + assert log.celery_task_id is None + assert [path for path, _ in billing.requests] == ["/quota/reserve", "/quota/release"] + assert all(not session.in_transaction() for session in harness.metadata.sessions) + assert billing.open_transactions + assert not any(billing.open_transactions) + + +@pytest.mark.parametrize("missing", ["app", "tenant", "workflow", "selected-workflow", "default-workflow"]) +def test_missing_or_foreign_ownership_is_rejected_before_billing( + harness: _Harness, + dispatch: _Dispatch, + billing: _Billing, + missing: str, +) -> None: + app_id, tenant_id, workflow_id = dispatch.app_id, harness.tenant_id, None + if missing == "app": + app_id = str(uuid4()) + elif missing == "tenant": + tenant_id = str(uuid4()) + elif missing == "workflow": + workflow_id = str(uuid4()) + else: + with harness.session_factory.begin() as session: + app = session.get(App, dispatch.app_id) + assert app is not None + if missing == "default-workflow": + app.workflow_id = None + else: + workflow_id = app.workflow_id + workflow = session.get(Workflow, workflow_id) + assert workflow is not None + workflow.app_id = str(uuid4()) + with pytest.raises(WorkflowNotFoundError): + _run(harness, dispatch, app_id=app_id, tenant_id=tenant_id, workflow_id=workflow_id) + assert billing.requests == [] + with harness.session_factory() as session: + assert session.scalar(select(WorkflowTriggerLog)) is None + + +def test_explicit_draft_workflow_preserves_domain_error( + harness: _Harness, + dispatch: _Dispatch, + billing: _Billing, +) -> None: + with harness.session_factory.begin() as session: + workflow = session.scalar(select(Workflow).where(Workflow.app_id == dispatch.app_id)) + assert workflow is not None + workflow.version = Workflow.VERSION_DRAFT + workflow_id = workflow.id + with pytest.raises(IsDraftWorkflowError, match="Cannot use draft workflow version"): + _run(harness, dispatch, workflow_id=workflow_id) + assert billing.requests == [] diff --git a/api/tests/unit_tests/services/test_webhook_service.py b/api/tests/unit_tests/services/test_webhook_service.py index ac7392fa303bb8..cffc3caaaecddf 100644 --- a/api/tests/unit_tests/services/test_webhook_service.py +++ b/api/tests/unit_tests/services/test_webhook_service.py @@ -9,7 +9,7 @@ from sqlalchemy.orm import Session, sessionmaker from werkzeug.datastructures import FileStorage -from models.enums import AppTriggerStatus, AppTriggerType, EndUserType +from models.enums import AppTriggerStatus, AppTriggerType, CreatorUserRole, EndUserType from models.model import App, AppMode, EndUser from models.tools import ToolFile from models.trigger import AppTrigger, WorkflowWebhookTrigger @@ -19,6 +19,7 @@ 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 services.workflow.entities import AsyncTriggerResponse, TriggerActor, TriggerData from tests.unit_tests.model_factories import make_app, make_end_user, make_workflow @@ -233,103 +234,79 @@ def test_published_lookup_reports_rate_limited_trigger( WebhookService.get_webhook_trigger_and_workflow("webhook-123") +class _WorkflowDispatcher: + def __init__(self, error: Exception | None = None) -> None: + self.error = error + self.actor: TriggerActor | None = None + self.data: TriggerData | None = None + + def trigger_workflow_async(self, *, actor: TriggerActor, trigger_data: TriggerData) -> AsyncTriggerResponse: + self.actor, self.data = actor, trigger_data + if self.error is not None: + raise self.error + return AsyncTriggerResponse(workflow_trigger_log_id="log", task_id="task", status="queued", queue="workflow") + + class TestWebhookServiceUnit: """Webhook business-logic tests with isolated sessions where the service owns their lifecycle.""" def test_trigger_workflow_execution_propagates_quota_error_without_error_log( self, 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)) - webhook_trigger = _webhook_trigger() - workflow = _workflow() quota_charge = MagicMock() quota_error = QuotaExceededError(feature="workflow", tenant_id="tenant-123", required=1) - + dispatch = _WorkflowDispatcher(error=quota_error) caplog.set_level(logging.INFO) - with ( - patch("services.trigger.webhook_service.QuotaService.reserve", return_value=quota_charge), - patch( - "services.trigger.webhook_service.AsyncWorkflowService.trigger_workflow_async", - side_effect=quota_error, - ) as mock_trigger_workflow_async, - ): + with patch("services.trigger.webhook_service.QuotaService.reserve", return_value=quota_charge): with pytest.raises(QuotaExceededError) as exc_info: WebhookService.trigger_workflow_execution( - webhook_trigger, + _webhook_trigger(), {"body": {}, "headers": {}, "query_params": {}, "files": {}, "method": "POST"}, - workflow, + _workflow(), end_users=_EndUserServiceStub(_end_user()), + workflow_dispatch=dispatch, trigger_status=trigger_status, ) - assert exc_info.value is quota_error quota_charge.refund.assert_called_once_with() - session = mock_trigger_workflow_async.call_args.kwargs["session"] - assert isinstance(session, Session) - assert session.in_transaction() is False assert len(caplog.records) == 1 assert caplog.records[0].levelno == logging.INFO - assert caplog.records[0].message == ( - "Tenant tenant-123 quota exceeded for feature workflow, skipping webhook trigger webhook-123" - ) - 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() - workflow = _workflow() - end_user = _end_user() + def test_trigger_workflow_execution_success(self, trigger_status: AppTriggerService) -> None: quota_charge = MagicMock() - webhook_data = { - "method": "POST", - "headers": {"Authorization": "Bearer token"}, - "query_params": {"version": "1"}, - "body": {"message": "hello"}, - "files": {}, - } - - with ( - patch.object(webhook_service_module.QuotaService, "reserve", return_value=quota_charge), - patch.object( - webhook_service_module.AsyncWorkflowService, - "trigger_workflow_async", - ) as mock_trigger, - ): + dispatch = _WorkflowDispatcher() + end_user = _end_user() + with patch.object(webhook_service_module.QuotaService, "reserve", return_value=quota_charge): WebhookService.trigger_workflow_execution( - webhook_trigger, - webhook_data, - workflow, + _webhook_trigger(), + {"method": "POST", "headers": {}, "query_params": {}, "body": {}, "files": {}}, + _workflow(), end_users=_EndUserServiceStub(end_user), + workflow_dispatch=dispatch, trigger_status=trigger_status, ) - - call_session = mock_trigger.call_args.kwargs["session"] - assert call_session.get_bind() is sqlite_engine + assert dispatch.actor is not None + assert dispatch.actor.id == end_user.id + assert dispatch.actor.role == CreatorUserRole.END_USER + assert dispatch.data is not None + assert dispatch.data.workflow_id == _workflow().id quota_charge.commit.assert_called_once_with() quota_charge.refund.assert_not_called() 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": {}} - + dispatch = _WorkflowDispatcher() with pytest.raises(ValueError, match="Failed to create end user"): WebhookService.trigger_workflow_execution( - webhook_trigger, - webhook_data, - workflow, + _webhook_trigger(), + {"method": "POST", "headers": {}, "query_params": {}, "body": {}, "files": {}}, + _workflow(), end_users=_EndUserServiceStub(ValueError("Failed to create end user")), + workflow_dispatch=dispatch, trigger_status=trigger_status, ) + assert dispatch.actor is None def test_extract_webhook_data_json(self): """Test webhook data extraction from JSON request.""" 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 582ca337f3e66f..c8584633fe27ac 100644 --- a/api/tests/unit_tests/tasks/test_trigger_processing_tasks.py +++ b/api/tests/unit_tests/tasks/test_trigger_processing_tasks.py @@ -32,7 +32,6 @@ from services.trigger.subscription_query_service import TriggerSubscriptionRuntime from services.trigger.trigger_request_service import TriggerHttpRequestCachingService from tasks.async_workflow_tasks import execute_workflow_sandbox -from tasks.trigger_processing_tasks import dispatch_triggered_workflow from tasks.workflow_cfs_scheduler.entities import AsyncWorkflowQueue from tests.unit_tests.config_override import config_overrides_context from tests.unit_tests.controllers.console.workspace import test_trigger_oauth_client, test_trigger_subscription_builder @@ -49,17 +48,20 @@ class _Billing: requests: list[tuple[str, dict[str, object]]] = field(default_factory=list) exhausted: bool = False + open_transactions: list[bool] = field(default_factory=list) + plan: str = "sandbox" @pytest.fixture -def billing() -> Iterator[_Billing]: +def billing(harness: _Harness) -> Iterator[_Billing]: billing = _Billing() class Handler(BaseHTTPRequestHandler): def do_GET(self) -> None: + billing.open_transactions.append(any(session.in_transaction() for session in harness.metadata.sessions)) self.respond( { - "subscription": {"plan": "sandbox", "interval": "month", "education": False}, + "subscription": {"plan": billing.plan, "interval": "month", "education": False}, "members": {"size": 1, "limit": 10}, "apps": {"size": 1, "limit": 10}, "documents_upload_quota": {"size": 0, "limit": 10}, @@ -71,6 +73,7 @@ def do_GET(self) -> None: ) def do_POST(self) -> None: + billing.open_transactions.append(any(session.in_transaction() for session in harness.metadata.sessions)) payload = json.loads(self.rfile.read(int(self.headers["Content-Length"]))) billing.requests.append((self.path, payload)) if self.path == "/quota/reserve": @@ -126,15 +129,11 @@ def invoke() -> int: config_overrides_context(DEPLOYMENT_EDITION=DeploymentEdition.CLOUD), ): services = application_services() - return dispatch_triggered_workflow( + return services.workspaces.trigger_workflow_dispatch.dispatch_event( user_id="webhook-actor", subscription=self.subscription, event_name="push", request_id=self.request_id, - 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) @@ -304,6 +303,8 @@ def test_dispatch_commits_quota_and_publishes_actual_workflow_task( assert sent.headers["X-Plugin-ID"] == "langgenius/github" assert sent.headers["X-Plugin-Unique-Identifier"] == "langgenius/github:1.0.0@package-hash" assert all(not session.in_transaction() for session in harness.metadata.sessions) + assert billing.open_transactions + assert not any(billing.open_transactions) @pytest.mark.parametrize("outcome", ["cancelled", "invoke-error"]) @@ -341,3 +342,5 @@ def test_dispatch_refunds_quota_without_publishing_for_rejected_event( assert log.status == WorkflowTriggerStatus.FAILED assert "Repository no longer exists" in (log.error or "") assert run.error == log.error + assert billing.open_transactions + assert not any(billing.open_transactions)