perf(api): retrieve published workflows via app.workflow_id (#38153)

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}")
 

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,

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(

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
     ):

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,