Merge branch 'main' into 408-introduce-primary-runtime-for-project

Author: Knerio

app/graphql/mutations/namespaces/projects/assign_runtimes.rb

@@ -0,0 +1,32 @@
+# frozen_string_literal: true
+
+module Mutations
+  module Namespaces
+    module Projects
+      class AssignRuntimes < BaseMutation
+        description 'Assign runtimes to a project'
+
+        argument :namespace_project_id, Types::GlobalIdType[::NamespaceProject],
+                 description: 'ID of the project to assign runtimes to'
+        argument :runtime_ids, [Types::GlobalIdType[::Runtime]], description: 'The new runtimes assigned to the project'
+
+        field :namespace_project, Types::NamespaceProjectType, null: true,
+                                                               description: 'The updated project with assigned runtimes'
+
+        def resolve(namespace_project_id:, runtime_ids:)
+          namespace_project = SagittariusSchema.object_from_id(namespace_project_id)
+          runtimes = runtime_ids.map { |runtime_id| SagittariusSchema.object_from_id(runtime_id) }
+
+          return { namespace_project: nil, errors: [create_message_error('Invalid project')] } if namespace_project.nil?
+          return { namespace_project: nil, errors: [create_message_error('Invalid runtime')] } if runtimes.any?(&:nil?)
+
+          ::Namespaces::Projects::AssignRuntimesService.new(
+            current_authentication,
+            namespace_project,
+            runtimes
+          ).execute.to_mutation_response(success_key: :namespace_project)
+        end
+      end
+    end
+  end
+end

app/graphql/types/mutation_type.rb

@@ -10,6 +10,7 @@ class MutationType < Types::BaseObject
     mount_mutation Mutations::Namespaces::Members::AssignRoles
     mount_mutation Mutations::Namespaces::Members::Delete
     mount_mutation Mutations::Namespaces::Members::Invite
+    mount_mutation Mutations::Namespaces::Projects::AssignRuntimes
     mount_mutation Mutations::Namespaces::Projects::Create
     mount_mutation Mutations::Namespaces::Projects::Update
     mount_mutation Mutations::Namespaces::Projects::Delete

app/graphql/types/namespace_project_type.rb

@@ -8,6 +8,7 @@ class NamespaceProjectType < Types::BaseObject
 
     field :description, String, null: false, description: 'Description of the project'
     field :name, String, null: false, description: 'Name of the project'
+    field :runtimes, Types::RuntimeType.connection_type, null: false, description: 'Runtimes assigned to this project'
 
     field :namespace, Types::NamespaceType, null: false,
                                             description: 'The namespace where this project belongs to'

app/graphql/types/runtime_status_type.rb

@@ -0,0 +1,10 @@
+# frozen_string_literal: true
+
+module Types
+  class RuntimeStatusType < BaseEnum
+    description 'Represent all available types of statuses of a runtime'
+
+    value :CONNECTED, 'No problem with connection, everything works as expected', value: :connected
+    value :DISCONNECTED, 'The runtime is disconnected, cause unknown', value: :disconnected
+  end
+end

app/graphql/types/runtime_type.rb

@@ -11,6 +11,10 @@ class RuntimeType < Types::BaseObject
     field :flow_types, Types::FlowTypeType.connection_type, null: false, description: 'FlowTypes of the runtime'
     field :name, String, null: false, description: 'The name for the runtime'
     field :namespace, Types::NamespaceType, null: true, description: 'The parent namespace for the runtime'
+    field :projects, Types::NamespaceProjectType.connection_type, null: false,
+                                                                  description: 'Projects associated with the runtime'
+    field :status, Types::RuntimeStatusType, null: false, description: 'The status of the runtime'
+
     field :token, String, null: true, description: 'Token belonging to the runtime, only present on creation'
 
     id_field Runtime

app/grpc/concerns/grpc_stream_handler.rb

@@ -0,0 +1,112 @@
+# frozen_string_literal: true
+
+module GrpcStreamHandler
+  include Code0::ZeroTrack::Loggable
+  extend ActiveSupport::Concern
+
+  class_methods do
+    def grpc_stream(method)
+      define_method(method) do |_, call|
+        current_runtime_id = Code0::ZeroTrack::Context.current[:runtime][:id]
+
+        create_enumerator(self.class, method, current_runtime_id, call.instance_variable_get(:@wrapped))
+      end
+
+      define_method("send_#{method}") do |grpc_object, runtime_id|
+        logger.info(message: 'Sending data', runtime_id: runtime_id, method: method)
+
+        encoded_data = self.class.encoders[method].call(grpc_object)
+        encoded_data64 = Base64.encode64(encoded_data)
+
+        logger.info(message: 'Encoded data', runtime_id: runtime_id, method: method, encoded_data: encoded_data64)
+
+        ActiveRecord::Base.connection.raw_connection
+                          .exec("NOTIFY grpc_streams, '#{self.class},#{method},#{runtime_id},#{encoded_data64}'")
+      end
+      define_method("end_#{method}") do |runtime_id|
+        ActiveRecord::Base.connection.raw_connection
+                          .exec("NOTIFY grpc_streams, '#{self.class},#{method},#{runtime_id},end'")
+      end
+    end
+  end
+
+  def self.stop_listen!
+    logger.info(message: 'Stopping listener')
+    GrpcStreamHandler.exiting = true
+  end
+
+  def self.listen!
+    ActiveRecord::Base.with_connection do |ar_conn|
+      conn = ar_conn.raw_connection
+      conn.exec('LISTEN grpc_streams')
+
+      logger.info(message: 'Listening for notifications on grpc_streams channel')
+      loop do
+        break if GrpcStreamHandler.exiting
+
+        conn.wait_for_notify(1) do |_, _, payload|
+          logger.info(message: 'Received notification', payload: payload)
+          class_name, method_name, runtime_id, encoded_data64 = payload.split(',')
+
+          clazz = class_name.constantize
+          method_name = method_name.to_sym
+
+          if encoded_data64 == 'end'
+            decoded_data = :end
+          else
+            data = Base64.decode64(encoded_data64)
+            decoded_data = clazz.decoders[method_name].call(data)
+          end
+
+          queues = GrpcStreamHandler.yielders.dig(clazz, method_name, runtime_id.to_i)
+          queues&.each do |queue|
+            queue << decoded_data
+          rescue StandardError => e
+            logger.error(message: 'Error while yielding data', error: e.message)
+          end
+        end
+      end
+      conn.exec('UNLISTEN grpc_streams')
+      logger.info(message: 'Stopped listening for notifications on grpc_streams channel')
+    end
+  end
+
+  def create_enumerator(clazz, method, runtime_id, _call)
+    logger.debug(message: 'Creating enumerator', runtime_id: runtime_id, clazz: clazz, method: method)
+
+    queue = Queue.new
+
+    enumerator = Enumerator.new do |y|
+      loop do
+        item = queue.pop(timeout: 1)
+        next if item.nil?
+        break if item == :end
+
+        begin
+          y << item
+        rescue GRPC::Core::CallError
+          logger.info(message: 'Stream was closed from client side (probably)')
+          clazz.try("#{method}_died", runtime_id)
+
+          raise
+        end
+      end
+      logger.info(message: 'Stream was closed from server side')
+      clazz.try("#{method}_died", runtime_id)
+    end
+
+    GrpcStreamHandler.yielders[clazz] ||= {}
+    GrpcStreamHandler.yielders[clazz][method] ||= {}
+    GrpcStreamHandler.yielders[clazz][method][runtime_id] ||= []
+
+    GrpcStreamHandler.yielders[clazz][method][runtime_id] << queue
+
+    clazz.try("#{method}_started", runtime_id)
+
+    enumerator
+  end
+
+  mattr_accessor :yielders, :exiting
+  GrpcStreamHandler.yielders = {}
+  GrpcStreamHandler.exiting = false
+end

app/grpc/flow_handler.rb

@@ -0,0 +1,24 @@
+# frozen_string_literal: true
+
+class FlowHandler < Tucana::Sagittarius::FlowService::Service
+  include GrpcHandler
+  include GrpcStreamHandler
+
+  grpc_stream :update
+
+  def self.update_started(runtime_id)
+    runtime = Runtime.find(runtime_id)
+    runtime.connected!
+    runtime.save
+  end
+
+  def self.update_died(runtime_id)
+    runtime = Runtime.find(runtime_id)
+    runtime.disconnected!
+    runtime.save
+  end
+
+  def self.encoders = { update: ->(grpc_object) { Tucana::Sagittarius::FlowResponse.encode(grpc_object) } }
+
+  def self.decoders = { update: ->(string) { Tucana::Sagittarius::FlowResponse.decode(string) } }
+end

app/models/audit_event.rb

@@ -31,6 +31,7 @@ class AuditEvent < ApplicationRecord
     user_identity_linked: 27,
     user_identity_unlinked: 28,
     attachment_updated: 29,
+    project_runtimes_assigned: 30,
   }.with_indifferent_access
 
   # rubocop:disable Lint/StructNewOverride

app/models/namespace_project.rb

@@ -4,6 +4,9 @@ class NamespaceProject < ApplicationRecord
   belongs_to :namespace, inverse_of: :projects
   belongs_to :primary_runtime, class_name: 'Runtime', optional: true
 
+  has_many :runtime_assignments, class_name: 'NamespaceProjectRuntimeAssignment', inverse_of: :namespace_project
+  has_many :runtimes, through: :runtime_assignments, inverse_of: :projects
+
   has_many :role_assignments, class_name: 'NamespaceRoleProjectAssignment',
                               inverse_of: :project
   has_many :assigned_roles, class_name: 'NamespaceRole', through: :role_assignments,

app/models/namespace_project_runtime_assignment.rb

@@ -0,0 +1,20 @@
+# frozen_string_literal: true
+
+class NamespaceProjectRuntimeAssignment < ApplicationRecord
+  belongs_to :runtime, inverse_of: :project_assignments
+  belongs_to :namespace_project, inverse_of: :runtime_assignments
+
+  validates :runtime, uniqueness: { scope: :namespace_project_id }
+
+  validate :validate_namespaces, if: :runtime_changed?
+  validate :validate_namespaces, if: :namespace_project_changed?
+
+  private
+
+  def validate_namespaces
+    return if runtime.namespace.nil?
+    return if runtime.namespace == namespace_project.namespace
+
+    errors.add(:runtime, 'must belong to the same namespace as the project')
+  end
+end

app/models/namespace_role_ability.rb

@@ -24,6 +24,7 @@ class NamespaceRoleAbility < ApplicationRecord
     delete_runtime: { db: 20, description: 'Allows to delete a runtime' },
     rotate_runtime_token: { db: 21, description: 'Allows to regenerate a runtime token' },
     assign_role_projects: { db: 22, description: 'Allows to change the assigned projects of a namespace role' },
+    assign_project_runtimes: { db: 23, description: 'Allows to assign runtimes to a project in the namespace' },
   }.with_indifferent_access
   enum :ability, ABILITIES.transform_values { |v| v[:db] }, prefix: :can
 

app/models/runtime.rb

@@ -7,6 +7,11 @@ class Runtime < ApplicationRecord
 
   token_attr :token, prefix: 's_rt_', length: 48
 
+  enum :status, { disconnected: 0, connected: 1 }, default: :disconnected
+
+  has_many :project_assignments, class_name: 'NamespaceProjectRuntimeAssignment', inverse_of: :runtime
+  has_many :projects, class_name: 'NamespaceProject', through: :project_assignments, inverse_of: :runtimes
+
   has_many :data_types, inverse_of: :runtime
 
   has_many :flow_types, inverse_of: :runtime

app/policies/namespace_project_policy.rb

@@ -9,6 +9,7 @@ class NamespaceProjectPolicy < BasePolicy
 
   rule { can_create_projects }.enable :read_namespace_project
 
+  customizable_permission :assign_project_runtimes
   customizable_permission :read_namespace_project
   customizable_permission :update_namespace_project
   customizable_permission :delete_namespace_project

app/services/namespaces/projects/assign_runtimes_service.rb

@@ -0,0 +1,50 @@
+# frozen_string_literal: true
+
+module Namespaces
+  module Projects
+    class AssignRuntimesService
+      include Sagittarius::Database::Transactional
+
+      attr_reader :current_authentication, :namespace_project, :runtimes
+
+      def initialize(current_authentication, namespace_project, runtimes)
+        @current_authentication = current_authentication
+        @namespace_project = namespace_project
+        @runtimes = runtimes
+      end
+
+      def execute
+        unless Ability.allowed?(current_authentication, :assign_project_runtimes, namespace_project)
+          return ServiceResponse.error(message: 'Missing permission', payload: :missing_permission)
+        end
+
+        transactional do |t|
+          old_assignments_for_audit_event = namespace_project.runtime_assignments.map do |assignment|
+            { id: assignment.runtime.id }
+          end
+
+          namespace_project.runtimes = runtimes
+          unless namespace_project.save
+            t.rollback_and_return! ServiceResponse.error(
+              message: 'Failed to assign runtimes to project',
+              payload: namespace_project.errors
+            )
+          end
+
+          AuditService.audit(
+            :project_runtimes_assigned,
+            author_id: current_authentication.user.id,
+            entity: namespace_project,
+            target: namespace_project,
+            details: {
+              new_runtimes: runtimes.map { |runtime| { id: runtime.id } },
+              old_runtimes: old_assignments_for_audit_event,
+            }
+          )
+
+          ServiceResponse.success(message: 'Assigned runtimes to a project', payload: namespace_project)
+        end
+      end
+    end
+  end
+end

bin/grpc_server

@@ -3,4 +3,4 @@
 
 require_relative '../config/environment'
 
-Sagittarius::Grpc::Launcher.new.run!
+Sagittarius::Grpc::Launcher.new.start_blocking

bin/test_grpc_client

@@ -0,0 +1,35 @@
+#!/usr/bin/env ruby
+
+require 'grpc'
+require 'tucana'
+
+Tucana.load_protocol(:sagittarius)
+
+def main
+  token = ARGV.size > 0 ? ARGV[0] : 'no-token-set'
+  hostname = ARGV.size > 1 ? ARGV[1] : 'localhost:50051'
+  stub = Tucana::Sagittarius::FlowService::Stub.new(hostname, :this_channel_is_insecure)
+  begin
+    message = Tucana::Sagittarius::FlowLogonRequest.new
+
+    puts "Sending: #{message.inspect}"
+
+    stub.update(message, {
+      metadata: {
+        authorization: token
+      }
+    }).each do |response|
+      puts "Received: #{response.inspect}"
+
+      puts "Simulating connection abort"
+      exit
+    end
+
+    puts "Done"
+
+  rescue GRPC::BadStatus => e
+    abort "ERROR: #{e.message}"
+  end
+end
+
+main

db/migrate/20250518095627_add_status_to_runtime.rb

@@ -0,0 +1,7 @@
+# frozen_string_literal: true
+
+class AddStatusToRuntime < Code0::ZeroTrack::Database::Migration[1.0]
+  def change
+    add_column :runtimes, :status, :integer, default: 0, null: false
+  end
+end

db/migrate/20250523224333_create_namespace_project_runtime_assignments.rb

@@ -0,0 +1,14 @@
+# frozen_string_literal: true
+
+class CreateNamespaceProjectRuntimeAssignments < Code0::ZeroTrack::Database::Migration[1.0]
+  def change
+    create_table :namespace_project_runtime_assignments do |t|
+      t.references :runtime, null: false, foreign_key: { on_delete: :cascade }, index: false
+      t.references :namespace_project, null: false, foreign_key: { on_delete: :cascade }, index: false
+
+      t.index %i[runtime_id namespace_project_id], unique: true
+
+      t.timestamps_with_timezone
+    end
+  end
+end

db/schema_migrations/20250518095627

@@ -0,0 +1 @@
+1ea1a340d1ae5c747ca07b3b45afef0b458b157a7118277eaf44b74dc0c03b91

db/schema_migrations/20250523224333

@@ -0,0 +1 @@
+4c06be8a4a4a60500cc1772d3c8066c809659ff4ff80441d0fd444c6c4bccf48