From cb35c6fa98f9c609210d38b0c9191055c4610ea2 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E6=9E=97=E7=8E=AE=20=28Jade=20Lin=29?= Date: Mon, 29 Jun 2026 17:51:21 +0800 Subject: [PATCH] perf(api): retrieve published workflows via app.workflow_id (#38153) --- api/services/trigger/webhook_service.py | 17 +++++---- api/tasks/trigger_processing_tasks.py | 36 ++++++++++--------- .../services/test_webhook_service.py | 3 ++ .../test_webhook_service_relationships.py | 35 ++++++++++++++++++ .../tasks/test_trigger_processing_tasks.py | 2 +- 5 files changed, 69 insertions(+), 24 deletions(-) diff --git a/api/services/trigger/webhook_service.py b/api/services/trigger/webhook_service.py index ac7f6a468eb..23b3ac55b93 100644 --- a/api/services/trigger/webhook_service.py +++ b/api/services/trigger/webhook_service.py @@ -40,6 +40,7 @@ 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_service import WorkflowService try: import magic @@ -114,6 +115,7 @@ class WebhookService: workflow = session.scalar( select(Workflow) .where( + Workflow.tenant_id == webhook_trigger.tenant_id, Workflow.app_id == webhook_trigger.app_id, Workflow.version == Workflow.VERSION_DRAFT, ) @@ -125,6 +127,7 @@ class WebhookService: app_trigger = session.scalar( select(AppTrigger) .where( + AppTrigger.tenant_id == webhook_trigger.tenant_id, AppTrigger.app_id == webhook_trigger.app_id, AppTrigger.node_id == webhook_trigger.node_id, AppTrigger.trigger_type == AppTriggerType.TRIGGER_WEBHOOK, @@ -145,16 +148,18 @@ class WebhookService: if app_trigger.status != AppTriggerStatus.ENABLED: raise ValueError(f"Webhook trigger is disabled for webhook {webhook_id}") - # Get workflow - workflow = session.scalar( - select(Workflow) + app = session.scalar( + select(App) .where( - Workflow.app_id == webhook_trigger.app_id, - Workflow.version != Workflow.VERSION_DRAFT, + App.tenant_id == webhook_trigger.tenant_id, + App.id == webhook_trigger.app_id, ) - .order_by(Workflow.created_at.desc()) .limit(1) ) + if not app: + raise ValueError(f"App not found for webhook {webhook_id}") + + workflow = WorkflowService().get_published_workflow(app, session=session) if not workflow: raise ValueError(f"Workflow not found for app {webhook_trigger.app_id}") diff --git a/api/tasks/trigger_processing_tasks.py b/api/tasks/trigger_processing_tasks.py index 4bd45a42ead..93f7f01407b 100644 --- a/api/tasks/trigger_processing_tasks.py +++ b/api/tasks/trigger_processing_tasks.py @@ -12,7 +12,7 @@ from datetime import UTC, datetime from typing import Any from celery import shared_task -from sqlalchemy import func, select +from sqlalchemy import select from sqlalchemy.orm import Session from core.db.session_factory import session_factory @@ -35,7 +35,7 @@ from models.enums import ( WorkflowRunTriggeredFrom, WorkflowTriggerStatus, ) -from models.model import EndUser +from models.model import App, EndUser from models.provider_ids import TriggerProviderID from models.trigger import TriggerSubscription, WorkflowPluginTrigger, WorkflowTriggerLog from models.workflow import Workflow, WorkflowAppLog, WorkflowAppLogCreatedFrom, WorkflowRun @@ -99,23 +99,25 @@ def dispatch_trigger_debug_event( return 0 -def _get_latest_workflows_by_app_ids( +def _get_published_workflows_by_app_ids( session: Session, subscribers: Sequence[WorkflowPluginTrigger] ) -> Mapping[str, Workflow]: - """Get the latest workflows by app_ids""" - workflow_query = ( - select(Workflow.app_id, func.max(Workflow.created_at).label("max_created_at")) - .where( - Workflow.app_id.in_({t.app_id for t in subscribers}), - Workflow.version != Workflow.VERSION_DRAFT, - ) - .group_by(Workflow.app_id) - .subquery() - ) + """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( - workflow_query, - (Workflow.app_id == workflow_query.c.app_id) & (Workflow.created_at == workflow_query.c.max_created_at), + 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} @@ -262,7 +264,7 @@ def dispatch_triggered_workflow( # Ensure expire_on_commit is set to False to remain workflows available with session_factory.create_session() as session: - workflows: Mapping[str, Workflow] = _get_latest_workflows_by_app_ids(session, subscribers) + workflows: Mapping[str, Workflow] = _get_published_workflows_by_app_ids(session, subscribers) end_users: Mapping[str, EndUser] = EndUserService.create_end_user_batch( type=EndUserType.TRIGGER, diff --git a/api/tests/test_containers_integration_tests/services/test_webhook_service.py b/api/tests/test_containers_integration_tests/services/test_webhook_service.py index a7e29045e0d..ae99cc05ffb 100644 --- a/api/tests/test_containers_integration_tests/services/test_webhook_service.py +++ b/api/tests/test_containers_integration_tests/services/test_webhook_service.py @@ -127,6 +127,9 @@ class TestWebhookService: db_session_with_containers.add(workflow) db_session_with_containers.flush() + app.workflow_id = workflow.id + db_session_with_containers.flush() + # Create webhook trigger webhook_id = fake.uuid4()[:16] webhook_trigger = WorkflowWebhookTrigger( 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 f37dc328b02..c699d39dde1 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 @@ -2,6 +2,7 @@ from __future__ import annotations import json import logging +from datetime import datetime from types import SimpleNamespace from unittest.mock import MagicMock, patch from uuid import uuid4 @@ -240,6 +241,40 @@ class TestWebhookServiceLookupWithContainers: with pytest.raises(ValueError, match="Workflow not found"): WebhookService.get_webhook_trigger_and_workflow(webhook_trigger.webhook_id) + def test_get_webhook_trigger_and_workflow_uses_app_workflow_id( + self, db_session_with_containers: Session, flask_app_with_containers: Flask + ): + del flask_app_with_containers + factory = WebhookServiceRelationshipFactory + account, tenant = factory.create_account_and_tenant(db_session_with_containers) + app = factory.create_app(db_session_with_containers, tenant, account) + current_workflow = factory.create_workflow( + db_session_with_containers, app=app, account=account, node_ids=["node-1"], version="2026-04-14.001" + ) + newer_workflow = factory.create_workflow( + db_session_with_containers, app=app, account=account, node_ids=["node-1"], version="2026-04-15.001" + ) + current_workflow.created_at = datetime(2026, 4, 14) + newer_workflow.created_at = datetime(2026, 4, 15) + app.workflow_id = current_workflow.id + db_session_with_containers.commit() + + webhook_trigger = factory.create_webhook_trigger( + db_session_with_containers, app=app, account=account, node_id="node-1" + ) + factory.create_app_trigger( + db_session_with_containers, app=app, node_id="node-1", status=AppTriggerStatus.ENABLED + ) + + got_trigger, got_workflow, got_node_config = WebhookService.get_webhook_trigger_and_workflow( + webhook_trigger.webhook_id + ) + + assert got_trigger.id == webhook_trigger.id + assert got_workflow.id == current_workflow.id + assert got_workflow.id != newer_workflow.id + assert got_node_config["id"] == "node-1" + def test_get_webhook_trigger_and_workflow_returns_debug_draft_workflow( self, db_session_with_containers: Session, flask_app_with_containers: Flask ): 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 59da5cc7a2a..cd5df4466e3 100644 --- a/api/tests/unit_tests/tasks/test_trigger_processing_tasks.py +++ b/api/tests/unit_tests/tasks/test_trigger_processing_tasks.py @@ -98,7 +98,7 @@ class TestDispatchTriggeredWorkflow: ), patch.object( trigger_processing_tasks_module, - "_get_latest_workflows_by_app_ids", + "_get_published_workflows_by_app_ids", ) as get_workflows, patch.object( trigger_processing_tasks_module.EndUserService,