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,27 @@
+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 disabled_authorization?
+
+      JWT.decode(params[:waiting_room_session_token], ENV.fetch('WAITING_ROOM_SIGNATURE'), true, { algorithm: 'HS256' })
+    end
+
+    def disabled_authorization?
+      ENV['DISABLE_WAITING_ROOM'] == 'yes'
+    end
+
+    def handle_waiting_room_session_token_error
+      render_error_payload(exception.message, 400)
+    end
+  end
+end

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

@@ -0,0 +1,23 @@
+# We use lambda instead of following API to generate token. Following are just sample.
+#
+# POST: /api/v2/storefront/waiting_room_sesions
+# 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,
+              firebase_document_id: params[:firebase_document_id]
+            )
+            render json: { token: context.token }
+          end
+        end
+      end
+    end
+  end
+end

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_refresh_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_refresh_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_refresh_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_refresh_to_enter_room_at] = Time.utc.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,44 @@
+require 'google/cloud/firestore'
+
+module SpreeCmCommissioner
+  class WaitingRoomSessionCreator < BaseInteractor
+    delegate :remote_ip, :firebase_document_id, to: :context
+
+    def call
+      return context.fail!(message: 'must_provide_firebase_document_id') if firebase_document_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
+      context.room_sesion = SpreeCmCommissioner::WaitingRoomSession.create!(token: context.jwt_token, expired_at: expired_at, remote_ip: remote_ip)
+    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(firebase_document_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_room_sessions_releaser_job.rb

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

app/models/spree_cm_commissioner/server_instance.rb

@@ -0,0 +1,20 @@
+module SpreeCmCommissioner
+  class ServerInstance < Base
+    validates :instance_id, presence: true
+    validates :concurrent_requests, presence: true
+
+    enum status: { active: 0, inactive: 1 }
+
+    def self.max_sessions_count
+      ServerInstance.active.select('SUM(concurrent_requests * multiply_by) AS total').take.total || 0
+    end
+
+    def self.min_session_count
+      ENV.fetch('MIN_SESSION_COUNT', '5').to_i
+    end
+
+    def self.max_sessions_count_with_min
+      [max_sessions_count, min_session_count].min
+    end
+  end
+end

app/models/spree_cm_commissioner/waiting_room_session.rb

@@ -0,0 +1,13 @@
+module SpreeCmCommissioner
+  class WaitingRoomSession < Base
+    validates :token, presence: true
+    validates :expired_at, presence: true
+    validates :ip, presence: true
+
+    scope :active, -> { SpreeCmCommissioner::WaitingRoomSession.where('expired_at > ?', Time.current) }
+
+    def expired?
+      expired_at < Time.current
+    end
+  end
+end

db/migrate/20241119032456_create_cm_waiting_room_sessions.rb

@@ -0,0 +1,14 @@
+class CreateCmWaitingRoomSessions < ActiveRecord::Migration[7.0]
+  def change
+    create_table :cm_waiting_room_sessions, if_not_exists: true do |t|
+      t.string :token, null: false
+      t.string :remote_ip
+      t.datetime :expired_at, null: false
+
+      t.timestamps
+    end
+
+    add_index :cm_waiting_room_sessions, :expired_at, if_not_exists: true
+    add_index :cm_waiting_room_sessions, :token, if_not_exists: true
+  end
+end

db/migrate/20241119090545_create_cm_server_instances.rb

@@ -0,0 +1,15 @@
+class CreateCmServerInstances < ActiveRecord::Migration[7.0]
+  def change
+    create_table :cm_server_instances, if_not_exists: true do |t|
+      t.string :instance_id, null: false
+      t.integer :concurrent_requests, default: 5, null: false
+      t.integer :multiply_by, default: 1, null: false
+      t.integer :status, default: 0, null: false
+
+      t.timestamps
+    end
+
+    add_index :cm_server_instances, :instance_id, if_not_exists: true
+    add_index :cm_server_instances, :status, if_not_exists: true
+  end
+end

lib/tasks/server_instances_bulk_creator.rake

@@ -0,0 +1,7 @@
+# recommend to be used in schedule.yml & manually access in /sidekiq/cron
+namespace :spree_cm_commissioner do
+  desc 'Create services instances by fetch from AWS'
+  task server_instances_bulk_creator: :environment do
+    ServerInstancesBulkCreatorJob.perform_now
+  end
+end