correct flow interface implementation

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,

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

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

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,

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

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

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

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

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

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