close #1991 add virtual waiting room

Gemfile.lock

@@ -37,12 +37,14 @@ PATH
     spree_cm_commissioner (0.0.1)
       activerecord_json_validator (~> 2.1, >= 2.1.3)
       aws-sdk-cloudfront
+      aws-sdk-ecs
       aws-sdk-s3
       byebug
       dry-validation (~> 1.10)
       elasticsearch (~> 8.5)
       exception_notification
       font-awesome-sass (~> 6.4.0)
+      google-cloud-firestore
       google-cloud-recaptcha_enterprise
       googleauth
       interactor (~> 3.1)
@@ -190,6 +192,9 @@ GEM
       aws-partitions (~> 1, >= 1.651.0)
       aws-sigv4 (~> 1.8)
       jmespath (~> 1, >= 1.6.1)
+    aws-sdk-ecs (1.147.0)
+      aws-sdk-core (~> 3, >= 3.193.0)
+      aws-sigv4 (~> 1.1)
     aws-sdk-kms (1.72.0)
       aws-sdk-core (~> 3, >= 3.184.0)
       aws-sigv4 (~> 1.1)
@@ -205,6 +210,7 @@ GEM
       execjs (~> 2.0)
     base64 (0.1.1)
     bcrypt (3.1.19)
+    bigdecimal (3.1.8)
     bootstrap (4.6.2)
       autoprefixer-rails (>= 9.1.0)
       popper_js (>= 1.16.1, < 2)
@@ -382,6 +388,19 @@ GEM
     google-cloud-env (1.6.0)
       faraday (>= 0.17.3, < 3.0)
     google-cloud-errors (1.3.1)
+    google-cloud-firestore (2.16.0)
+      bigdecimal (~> 3.0)
+      concurrent-ruby (~> 1.0)
+      google-cloud-core (~> 1.5)
+      google-cloud-firestore-v1 (>= 0.10, < 2.a)
+      rbtree (~> 0.4.2)
+    google-cloud-firestore-v1 (0.10.0)
+      gapic-common (>= 0.17.1, < 2.a)
+      google-cloud-errors (~> 1.0)
+      google-cloud-location (>= 0.4, < 2.a)
+    google-cloud-location (0.6.0)
+      gapic-common (>= 0.20.0, < 2.a)
+      google-cloud-errors (~> 1.0)
     google-cloud-recaptcha_enterprise (1.3.0)
       google-cloud-core (~> 1.6)
       google-cloud-recaptcha_enterprise-v1 (>= 0.0, < 2.a)
@@ -605,6 +624,7 @@ GEM
       activerecord (>= 6.0.4)
       activesupport (>= 6.0.4)
       i18n
+    rbtree (0.4.6)
     redis (5.0.7)
       redis-client (>= 0.9.0)
     redis-client (0.17.0)

app/controllers/concerns/spree_cm_commissioner/waiting_room_authorization.rb

@@ -0,0 +1,28 @@
+module SpreeCmCommissioner
+  module WaitingRoomAuthorization
+    extend ActiveSupport::Concern
+
+    included do
+      rescue_from JWT::ExpiredSignature, with: :handle_waiting_room_session_token_error
+      rescue_from JWT::VerificationError, with: :handle_waiting_room_session_token_error
+      rescue_from JWT::DecodeError, with: :handle_waiting_room_session_token_error
+
+      before_action :ensure_waiting_room_session_token
+    end
+
+    def ensure_waiting_room_session_token
+      return if ENV['DISABLE_WAITING_ROOM'] == 'yes'
+      return unless required_waiting_room_session?
+
+      JWT.decode(params[:waiting_room_session_token], ENV.fetch('WAITING_ROOM_SIGNATURE'), true, { algorithm: 'HS256' })
+    end
+
+    def required_waiting_room_session?
+      false
+    end
+
+    def handle_waiting_room_session_token_error
+      render_error_payload('invalid_session_token', 400)
+    end
+  end
+end

app/controllers/spree/admin/server_instances_controller.rb

@@ -0,0 +1,54 @@
+module Spree
+  module Admin
+    class ServerInstancesController < Spree::Admin::ResourceController
+      def index
+        @max_sessions_count = SpreeCmCommissioner::ServerInstance.max_sessions_count_with_min
+        @active_sesions_count = SpreeCmCommissioner::WaitingRoomSession.active.count
+      end
+
+      def collection
+        model_class.order(:id)
+      end
+
+      def modify_multiply_by
+        modifier = params[:multiply_by]&.to_i || 0
+
+        object = model_class.find(params[:id])
+        object.update(multiply_by: [object.multiply_by + modifier, 1].max)
+
+        redirect_back fallback_location: collection_url
+      end
+
+      def modify_concurrent_requests
+        modifier = params[:concurrent_requests]&.to_i || 0
+
+        object = model_class.find(params[:id])
+        object.update(concurrent_requests: [object.concurrent_requests + modifier, 1].max)
+
+        redirect_back fallback_location: collection_url
+      end
+
+      def force_pull
+        SpreeCmCommissioner::ServerInstancesBulkCreatorJob.perform_now
+        SpreeCmCommissioner::WaitingGuestsCallerJob.perform_now
+
+        redirect_back fallback_location: collection_url
+      end
+
+      # override
+      def collection_url(options = {})
+        admin_server_instances_url(options)
+      end
+
+      # override
+      def object_name
+        'spree_cm_commissioner_server_instance'
+      end
+
+      # override
+      def model_class
+        SpreeCmCommissioner::ServerInstance
+      end
+    end
+  end
+end

app/controllers/spree/api/v2/resource_controller_decorator.rb

@@ -3,6 +3,8 @@ module Api
     module V2
       module ResourceControllerDecorator
         def self.prepended(base)
+          base.include SpreeCmCommissioner::WaitingRoomAuthorization
+
           base.before_action :required_schema_validation!
           base.rescue_from SpreeCmCommissioner::SchemaValidationError, with: :rescue_schema_validation_error
           base.attr_reader :schema

app/controllers/spree/api/v2/storefront/accommodations_controller.rb

@@ -3,6 +3,11 @@ module Api
     module V2
       module Storefront
         class AccommodationsController < ::Spree::Api::V2::ResourceController
+          # overrided
+          def required_waiting_room_session?
+            true
+          end
+
           private
 
           def collection

app/controllers/spree/api/v2/storefront/homepage_sections_controller.rb

@@ -3,6 +3,11 @@ module Api
     module V2
       module Storefront
         class HomepageSectionsController < ::Spree::Api::V2::ResourceController
+          # overrided
+          def required_waiting_room_session?
+            true
+          end
+
           def model_class
             SpreeCmCommissioner::HomepageSection
           end

app/controllers/spree/api/v2/storefront/waiting_room_sessions_controller.rb

@@ -0,0 +1,32 @@
+# We use lambda instead of following API to generate token. Following are just sample.
+#
+# POST: /api/v2/storefront/waiting_room_sessions
+# Response: { token: token }
+module Spree
+  module Api
+    module V2
+      module Storefront
+        class WaitingRoomSessionsController < ::Spree::Api::V2::ResourceController
+          skip_before_action :ensure_waiting_room_session_token
+
+          def create
+            context = SpreeCmCommissioner::WaitingRoomSessionCreator.call(
+              remote_ip: request.remote_ip,
+              waiting_guest_firebase_doc_id: params[:waiting_guest_firebase_doc_id]
+            )
+
+            if context.success?
+              render_serialized_payload { serialize_resource(context.room_session) }
+            else
+              render_error_payload(context.message)
+            end
+          end
+
+          def resource_serializer
+            SpreeCmCommissioner::V2::Storefront::WaitingRoomSessionSerializer
+          end
+        end
+      end
+    end
+  end
+end

app/controllers/spree_cm_commissioner/api/v2/storefront/cart_controller_decorator.rb

@@ -7,6 +7,11 @@ def self.prepended(base)
             base.before_action :ensure_cart_exist, only: :add_item
           end
 
+          # overrided
+          def required_waiting_room_session?
+            true
+          end
+
           # one usecase where this neccessary is when order.state is 'complete', but complated_at & payment_state is null,
           # this cause app to be stuck at book page because cart is not considered cart or completed.
           #

app/interactors/spree_cm_commissioner/server_instances_bulk_creator.rb

@@ -0,0 +1,33 @@
+require 'aws-sdk-ecs'
+
+# TODO: UI to check all instances, max sessions, active_sessions, and available slots
+module SpreeCmCommissioner
+  class ServerInstancesBulkCreator < BaseInteractor
+    def call
+      ecs_client = Aws::ECS::Client.new(
+        region: ENV.fetch('AWS_REGION'),
+        access_key_id: ENV.fetch('AWS_ACCESS_KEY_ID'),
+        secret_access_key: ENV.fetch('AWS_SECRET_ACCESS_KEY')
+      )
+
+      tasks_response = ecs_client.list_tasks(
+        cluster: ENV.fetch('AWS_CLUSTER_NAME'),
+        launch_type: 'FARGATE'
+      )
+
+      context.servers = []
+      context.task_arns = tasks_response.task_arns
+      context.task_arns.each do |task_arn|
+        server = SpreeCmCommissioner::ServerInstance.where(instance_id: task_arn).first_or_initialize
+        server.concurrent_requests ||= 5
+        server.status = :active
+        server.save! if server.changed?
+        context.servers << server
+      end
+
+      SpreeCmCommissioner::ServerInstance.where.not(id: context.servers.pluck(:id)).find_each do |server|
+        server.update!(status: :inactive)
+      end
+    end
+  end
+end

app/interactors/spree_cm_commissioner/waiting_guests_caller.rb

@@ -0,0 +1,45 @@
+require 'google/cloud/firestore'
+
+# TODO: alert when available_slots is negative.
+module SpreeCmCommissioner
+  class WaitingGuestsCaller < BaseInteractor
+    def call
+      available_slots = fetch_available_slots
+      return unless available_slots.positive?
+
+      long_waiting_guests = fetch_long_waiting_guests(available_slots)
+      calling_all(long_waiting_guests)
+    end
+
+    def fetch_available_slots
+      max_sessions = SpreeCmCommissioner::ServerInstance.max_sessions_count_with_min
+      active_sessions = SpreeCmCommissioner::WaitingRoomSession.active.count
+      max_sessions - active_sessions
+    end
+
+    # This query required index. create them in Firebase beforehand.
+    # Client side must create waiting_guests document with :queued_at & :allow_to_enter_room_at to null to allow fillter & order.
+    def fetch_long_waiting_guests(available_slots)
+      firestore = Google::Cloud::Firestore.new(project_id: service_account[:project_id], credentials: service_account)
+      firestore.col('waiting_guests')
+               .where('allow_to_enter_room_at', '==', nil)
+               .order('queued_at')
+               .limit(available_slots)
+               .get
+    end
+
+    # For alert waiting guests to enter room, we just update :allow_to_enter_room_at.
+    # App will listen to firebase & start refresh session token to enter room.
+    def calling_all(waiting_guests)
+      waiting_guests.each do |document|
+        data = document.get.data.dup
+        data[:allow_to_enter_room_at] = Time.zone.now
+        document.update(data)
+      end
+    end
+
+    def service_account
+      Rails.application.credentials.cloud_firestore_service_account
+    end
+  end
+end

app/interactors/spree_cm_commissioner/waiting_room_session_creator.rb

@@ -0,0 +1,51 @@
+require 'google/cloud/firestore'
+
+module SpreeCmCommissioner
+  class WaitingRoomSessionCreator < BaseInteractor
+    delegate :remote_ip, :waiting_guest_firebase_doc_id, to: :context
+
+    def call
+      return context.fail!(message: 'must_provide_waiting_guest_firebase_doc_id') if waiting_guest_firebase_doc_id.blank?
+      return context.fail!(message: 'must_provide_remote_ip') if remote_ip.blank?
+
+      generate_jwt_token
+      assign_token_and_create_session_to_db
+      log_to_firebase
+    end
+
+    def generate_jwt_token
+      payload = { exp: expired_at.to_i }
+      context.jwt_token = JWT.encode(payload, ENV.fetch('WAITING_ROOM_SIGNATURE'), 'HS256')
+    end
+
+    def assign_token_and_create_session_to_db
+      # create or renew
+      context.room_session = SpreeCmCommissioner::WaitingRoomSession.where(guest_identifier: waiting_guest_firebase_doc_id).first_or_initialize
+      context.room_session.assign_attributes(
+        jwt_token: context.jwt_token,
+        expired_at: expired_at,
+        remote_ip: remote_ip
+      )
+      context.room_session.save!
+    end
+
+    def log_to_firebase
+      firestore = Google::Cloud::Firestore.new(project_id: service_account[:project_id], credentials: service_account)
+      document = firestore.col('waiting_guests').doc(waiting_guest_firebase_doc_id)
+
+      data = document.get.data.dup
+      data[:entered_room_at] = Time.zone.now
+
+      document.update(data)
+    end
+
+    def expired_at
+      expired_duration = ENV['WAITING_ROOM_SESSION_EXPIRE_DURATION_IN_SECOND']&.presence&.to_i || (60 * 5)
+      context.expired_at ||= expired_duration.seconds.from_now
+    end
+
+    def service_account
+      Rails.application.credentials.cloud_firestore_service_account
+    end
+  end
+end

app/jobs/spree_cm_commissioner/server_instances_bulk_creator_job.rb

@@ -0,0 +1,12 @@
+# server_instances_bulk_creator:
+#   cron: cron: "*/30 * * * *" # Every 30 minute
+#   class: SpreeCmCommissioner::ServerInstanceBulkCreatorJob
+module SpreeCmCommissioner
+  class ServerInstancesBulkCreatorJob < ApplicationJob
+    def perform
+      return if ENV['DISABLE_WAITING_ROOM'] == 'yes'
+
+      SpreeCmCommissioner::ServerInstancesBulkCreator.call
+    end
+  end
+end

app/jobs/spree_cm_commissioner/waiting_guests_caller_job.rb

@@ -0,0 +1,12 @@
+# waiting_guests_caller:
+#   cron: cron: "*/1 * * * *" # Every minute
+#   class: SpreeCmCommissioner::WaitingGuestsCallerJob
+module SpreeCmCommissioner
+  class WaitingGuestsCallerJob < ApplicationJob
+    def perform
+      return if ENV['DISABLE_WAITING_ROOM'] == 'yes'
+
+      SpreeCmCommissioner::WaitingGuestsCaller.call
+    end
+  end
+end