diff --git a/app/grpc/flow_handler.rb b/app/grpc/flow_handler.rb index 40abee55..d9e0a23a 100644 --- a/app/grpc/flow_handler.rb +++ b/app/grpc/flow_handler.rb @@ -7,6 +7,22 @@ class FlowHandler < Tucana::Sagittarius::FlowService::Service grpc_stream :update + def self.update_flow(flow) + response = Tucana::Sagittarius::FlowResponse.new(updated_flow: flow.to_grpc) + send_to_project_runtimes(flow.project, response) + end + + def self.delete_flow(project, flow_id) + response = Tucana::Sagittarius::FlowResponse.new(deleted_flow_id: flow_id) + send_to_project_runtimes(project, response) + end + + def self.send_to_project_runtimes(project, response) + project.runtime_assignments.compatible.find_each do |assignment| + send_update(response, assignment.runtime_id) + end + end + def self.update_runtime(runtime) assignments = runtime.project_assignments.compatible.includes( :namespace_project, diff --git a/app/jobs/delete_flow_for_project_job.rb b/app/jobs/delete_flow_for_project_job.rb new file mode 100644 index 00000000..c5b2e68d --- /dev/null +++ b/app/jobs/delete_flow_for_project_job.rb @@ -0,0 +1,10 @@ +# frozen_string_literal: true + +class DeleteFlowForProjectJob < ApplicationJob + def perform(project_id, flow_id) + project = NamespaceProject.find_by(id: project_id) + return if project.nil? + + FlowHandler.delete_flow(project, flow_id) + end +end diff --git a/app/jobs/update_flow_for_project_job.rb b/app/jobs/update_flow_for_project_job.rb new file mode 100644 index 00000000..a353c18d --- /dev/null +++ b/app/jobs/update_flow_for_project_job.rb @@ -0,0 +1,10 @@ +# frozen_string_literal: true + +class UpdateFlowForProjectJob < ApplicationJob + def perform(flow_id) + flow = Flow.find_by(id: flow_id) + return unless flow&.validation_status_valid? + + FlowHandler.update_flow(flow) + end +end diff --git a/app/services/namespaces/projects/flows/delete_service.rb b/app/services/namespaces/projects/flows/delete_service.rb index 423c7a41..ff81116d 100644 --- a/app/services/namespaces/projects/flows/delete_service.rb +++ b/app/services/namespaces/projects/flows/delete_service.rb @@ -29,7 +29,7 @@ def execute ) end - UpdateRuntimesForProjectJob.perform_later(flow.project.id) + DeleteFlowForProjectJob.perform_later(flow.project.id, flow.id) AuditService.audit( :flow_deleted, diff --git a/app/services/namespaces/projects/flows/validation_service.rb b/app/services/namespaces/projects/flows/validation_service.rb index f0de625d..09bf9f13 100644 --- a/app/services/namespaces/projects/flows/validation_service.rb +++ b/app/services/namespaces/projects/flows/validation_service.rb @@ -27,7 +27,7 @@ def execute validation_diagnostics: result.diagnostics ) - UpdateRuntimesForProjectJob.perform_later(flow.project.id) + UpdateFlowForProjectJob.perform_later(flow.id) if result.valid? result end diff --git a/spec/grpc/flow_handler_spec.rb b/spec/grpc/flow_handler_spec.rb index 5f28565b..30ae48fa 100644 --- a/spec/grpc/flow_handler_spec.rb +++ b/spec/grpc/flow_handler_spec.rb @@ -55,4 +55,43 @@ ) end end + + describe 'project runtime updates' do + let(:flow) { create(:flow, validation_status: :valid) } + let(:runtime) { create(:runtime, namespace: flow.project.namespace) } + + before do + create( + :namespace_project_runtime_assignment, + namespace_project: flow.project, + runtime: runtime, + compatible: true + ) + allow(described_class).to receive(:send_update) + end + + describe '.update_flow' do + it 'sends the updated_flow response to compatible project runtimes' do + described_class.update_flow(flow) + + expect(described_class).to have_received(:send_update) do |response, runtime_id| + expect(runtime_id).to eq(runtime.id) + expect(response.data).to eq(:updated_flow) + expect(response.updated_flow).to eq(flow.to_grpc) + end + end + end + + describe '.delete_flow' do + it 'sends the deleted_flow_id response to compatible project runtimes' do + described_class.delete_flow(flow.project, flow.id) + + expect(described_class).to have_received(:send_update) do |response, runtime_id| + expect(runtime_id).to eq(runtime.id) + expect(response.data).to eq(:deleted_flow_id) + expect(response.deleted_flow_id).to eq(flow.id) + end + end + end + end end diff --git a/spec/jobs/delete_flow_for_project_job_spec.rb b/spec/jobs/delete_flow_for_project_job_spec.rb new file mode 100644 index 00000000..def927cf --- /dev/null +++ b/spec/jobs/delete_flow_for_project_job_spec.rb @@ -0,0 +1,17 @@ +# frozen_string_literal: true + +require 'rails_helper' + +RSpec.describe DeleteFlowForProjectJob do + include ActiveJob::TestHelper + + it 'sends the deleted flow id' do + project = create(:namespace_project) + flow_id = 123 + allow(FlowHandler).to receive(:delete_flow) + + perform_enqueued_jobs { described_class.perform_later(project.id, flow_id) } + + expect(FlowHandler).to have_received(:delete_flow).with(project, flow_id) + end +end diff --git a/spec/jobs/update_flow_for_project_job_spec.rb b/spec/jobs/update_flow_for_project_job_spec.rb new file mode 100644 index 00000000..1b8e97b8 --- /dev/null +++ b/spec/jobs/update_flow_for_project_job_spec.rb @@ -0,0 +1,29 @@ +# frozen_string_literal: true + +require 'rails_helper' + +RSpec.describe UpdateFlowForProjectJob do + include ActiveJob::TestHelper + + let(:flow) { create(:flow) } + + before do + allow(FlowHandler).to receive(:update_flow) + end + + it 'sends a valid flow update' do + flow.update!(validation_status: :valid) + + perform_enqueued_jobs { described_class.perform_later(flow.id) } + + expect(FlowHandler).to have_received(:update_flow).with(flow) + end + + it 'does not send a deletion for a newly created invalid flow' do + flow.update!(validation_status: :invalid) + + perform_enqueued_jobs { described_class.perform_later(flow.id) } + + expect(FlowHandler).not_to have_received(:update_flow) + end +end diff --git a/spec/services/namespaces/projects/flows/delete_service_spec.rb b/spec/services/namespaces/projects/flows/delete_service_spec.rb index a6b38fe0..c98a788d 100644 --- a/spec/services/namespaces/projects/flows/delete_service_spec.rb +++ b/spec/services/namespaces/projects/flows/delete_service_spec.rb @@ -60,11 +60,11 @@ end it 'queues job to update runtimes' do - allow(UpdateRuntimesForProjectJob).to receive(:perform_later) + allow(DeleteFlowForProjectJob).to receive(:perform_later) service_response - expect(UpdateRuntimesForProjectJob).to have_received(:perform_later).with(namespace_project.id) + expect(DeleteFlowForProjectJob).to have_received(:perform_later).with(namespace_project.id, flow.id) end end end diff --git a/spec/services/namespaces/projects/flows/validation_service_spec.rb b/spec/services/namespaces/projects/flows/validation_service_spec.rb index 48295e29..ff5a5309 100644 --- a/spec/services/namespaces/projects/flows/validation_service_spec.rb +++ b/spec/services/namespaces/projects/flows/validation_service_spec.rb @@ -46,7 +46,7 @@ context 'with fixed validation results' do before do flow.update!(starting_node: node_function) - allow(UpdateRuntimesForProjectJob).to receive(:perform_later) + allow(UpdateFlowForProjectJob).to receive(:perform_later) result = Triangulum::Validation::Result.new(valid?: valid, return_type: nil, diagnostics: diagnostics) allow(Triangulum::Validation).to receive(:new).and_return( @@ -71,10 +71,10 @@ expect(flow.reload.validation_diagnostics).to eq([]) end - it 'enqueues UpdateRuntimesForProjectJob' do + it 'enqueues UpdateFlowForProjectJob' do service.execute - expect(UpdateRuntimesForProjectJob).to have_received(:perform_later).with(namespace_project.id) + expect(UpdateFlowForProjectJob).to have_received(:perform_later).with(flow.id) end end @@ -128,10 +128,18 @@ ) end - it 'enqueues UpdateRuntimesForProjectJob' do + it 'does not enqueue a runtime update for a newly created invalid flow' do service.execute - expect(UpdateRuntimesForProjectJob).to have_received(:perform_later).with(namespace_project.id) + expect(UpdateFlowForProjectJob).not_to have_received(:perform_later) + end + + it 'does not enqueue a deletion when a previously valid flow becomes invalid' do + flow.update!(validation_status: :valid) + + service.execute + + expect(UpdateFlowForProjectJob).not_to have_received(:perform_later) end end end