Merge pull request #1025 from code0-tech/fix-execution-issues

Fix subscription for execution results

Author: Taucher2003

app/graphql/subscription_triggers.rb

@@ -5,7 +5,8 @@ def self.execution_result(execution_result)
     SagittariusSchema.subscriptions.trigger(
       :namespaces_projects_flows_execution_result,
       { execution_identifier: execution_result.execution_identifier },
-      execution_result
+      execution_result,
+      context: { visibility_profile: :execution }
     )
   end
 end

app/graphql/subscriptions/base_subscription.rb

@@ -6,14 +6,19 @@ class BaseSubscription < GraphQL::Schema::Subscription
     field_class Types::BaseField
     object_class Types::BaseObject
 
-    def current_authentication
-      context[:current_authentication]
+    def self.inherited(subclass)
+      super
+      subclass.graphql_name subclass.name.delete_prefix('Subscriptions::').gsub('::', '')
     end
 
     def self.generate_payload_type
       result = super
       result.graphql_name("#{graphql_name}SubscriptionPayload")
       result
     end
+
+    def current_authentication
+      context[:current_authentication]
+    end
   end
 end

…s/flows/execution_result_subscription.rb …paces/projects/flows/execution_result.rbapp/graphql/subscriptions/namespaces/projects/flows/execution_result_subscription.rb renamed to app/graphql/subscriptions/namespaces/projects/flows/execution_result.rb app/graphql/subscriptions/namespaces/projects/flows/execution_result_subscription.rb renamed to app/graphql/subscriptions/namespaces/projects/flows/execution_result.rb

@@ -4,7 +4,7 @@ module Subscriptions
   module Namespaces
     module Projects
       module Flows
-        class ExecutionResultSubscription < BaseSubscription
+        class ExecutionResult < BaseSubscription
           description 'Subscription to asynchronously receive an execution result'
 
           argument :execution_identifier,
@@ -18,7 +18,7 @@ class ExecutionResultSubscription < BaseSubscription
                 description: 'The execution result of the relevant execution'
 
           def subscribe(execution_identifier:)
-            result = ExecutionResult.find_by(execution_identifier: execution_identifier)
+            result = ::ExecutionResult.find_by(execution_identifier: execution_identifier)
 
             if result.present?
               unsubscribe({ execution_result: result })

app/graphql/types/subscription_type.rb

@@ -6,7 +6,7 @@ class SubscriptionType < Types::BaseObject
 
     include Sagittarius::Graphql::MountSubscription
 
-    mount_subscription Subscriptions::Namespaces::Projects::Flows::ExecutionResultSubscription
+    mount_subscription Subscriptions::Namespaces::Projects::Flows::ExecutionResult
     mount_subscription Subscriptions::Echo
   end
 end

app/grpc/execution_handler.rb

@@ -21,8 +21,8 @@ def test(requests, call)
     end
 
     Thread.new do
-      ActiveRecord::Base.connection_pool.with_connection do
-        requests.each do |request|
+      requests.each do |request|
+        ApplicationRecord.connection_pool.with_connection do
           case request.data
           when :logon
             logger.info(message: 'Execution runtime sent logon', runtime_id: current_runtime_id)

…scription/executionresultsubscription.md …amespacesprojectsflowsexecutionresult.mddocs/graphql/subscription/executionresultsubscription.md renamed to docs/graphql/subscription/namespacesprojectsflowsexecutionresult.md docs/graphql/subscription/executionresultsubscription.md renamed to docs/graphql/subscription/namespacesprojectsflowsexecutionresult.md

@@ -1,5 +1,5 @@
 ---
-title: executionResultSubscription
+title: namespacesProjectsFlowsExecutionResult
 ---
 
 Subscription to asynchronously receive an execution result

spec/requests/graphql/subscription/namespaces/projects/flows/execution_result_spec.rb

@@ -0,0 +1,48 @@
+# frozen_string_literal: true
+
+require 'rails_helper'
+
+RSpec.describe 'namespacesProjectsFlowsExecutionResult Subscription', type: :channel do
+  include AuthenticationHelpers
+  include ActionCable::Channel::TestCase::Behavior
+
+  include_context 'with graphql subscription support'
+
+  tests GraphqlChannel
+
+  let(:user) { create(:user) }
+  let(:token) { "Session #{authorization_token(user)}" }
+  let(:flow) { create(:flow) }
+  let(:execution_identifier) { 'existing-execution' }
+
+  let(:subscription_query) do
+    <<~GQL
+      subscription($executionIdentifier: String!) {
+        namespacesProjectsFlowsExecutionResult(executionIdentifier: $executionIdentifier) {
+          executionResult { success }
+        }
+      }
+    GQL
+  end
+
+  before do
+    create(:namespace_member, namespace: flow.project.namespace, user: user)
+    stub_allowed_ability(NamespaceProjectPolicy, :read_namespace_project, user: user, subject: flow.project)
+
+    subscribe(token: token)
+  end
+
+  context 'when the execution result already exists' do
+    before do
+      create(:execution_result, flow: flow, execution_identifier: execution_identifier, success: { 'done' => true })
+    end
+
+    it 'immediately delivers the result in the initial subscription response' do
+      perform :execute, query: subscription_query, variables: { executionIdentifier: execution_identifier }
+
+      result = transmissions.last
+      expect(result.dig('result', 'data', 'namespacesProjectsFlowsExecutionResult', 'executionResult', 'success'))
+        .to eq({ 'done' => true })
+    end
+  end
+end

spec/services/namespaces/projects/flows/persist_execution_result_service_spec.rb

@@ -204,4 +204,43 @@
       expect(service_response.payload[:error_code]).to eq(:invalid_execution_result)
     end
   end
+
+  context 'with subscription integration', type: :channel do
+    include AuthenticationHelpers
+    include ActionCable::Channel::TestCase::Behavior
+
+    include_context 'with graphql subscription support'
+
+    tests GraphqlChannel
+
+    let(:user) { create(:user) }
+    let(:token) { "Session #{authorization_token(user)}" }
+
+    before do
+      allow(SubscriptionTriggers).to receive(:execution_result).and_call_original
+
+      create(:namespace_member, namespace: flow.project.namespace, user: user)
+      stub_allowed_ability(NamespaceProjectPolicy, :read_namespace_project, user: user, subject: flow.project)
+
+      subscribe(token: token)
+
+      perform :execute,
+              query: <<~GQL,
+                subscription($executionIdentifier: String!) {
+                  namespacesProjectsFlowsExecutionResult(executionIdentifier: $executionIdentifier) {
+                    executionResult { success }
+                  }
+                }
+              GQL
+              variables: { executionIdentifier: 'execution-identifier' }
+    end
+
+    it 'delivers the execution result to subscribers without visibility profile error' do
+      service_response
+
+      result = transmissions.last
+      expect(result.dig('result', 'data', 'namespacesProjectsFlowsExecutionResult', 'executionResult', 'success'))
+        .to eq({ 'result' => true })
+    end
+  end
 end