Add Queue for Registering (#39)

Add Queue locally and in the cloud

.github/workflows/deploy.yaml

@@ -18,11 +18,24 @@ jobs:
       - name: Login to Amazon ECR
         id: login-ecr
         uses: aws-actions/amazon-ecr-login@v1
-      - name: Build and push images
+      - name: Build and push handler
         uses: docker/build-push-action@v4
         with:
           context: .
           push: true
-          tags: ${{ steps.login-ecr.outputs.registry }}/wca-registration:latest
+          file: dockerfile.handler
+          tags: ${{ steps.login-ecr.outputs.registry }}/wca-registration-handler:latest
           cache-from: type=gha
           cache-to: type=gha,mode=max
+      - name: Build and push worker
+        uses: docker/build-push-action@v4
+        with:
+          context: .
+          push: true
+          file: dockerfile.worker
+          tags: ${{ steps.login-ecr.outputs.registry }}/wca-registration-worker:latest
+          cache-from: type=gha
+          cache-to: type=gha,mode=max
+      - name: Deploy worker
+        run: |
+          aws ecs update-service --cluster wca-registration --service wca-registration-worker --force-new-deployment

.gitignore

@@ -35,5 +35,5 @@
 .idea
 tmp
 node_modules
-storage
+localstack
 yarn.lock

Gemfile

@@ -3,6 +3,7 @@ git_source(:github) { |repo| "https://github.com/thewca/wca-registration.git" }
 
 ruby "3.2.2"
 
+# Gems that are only needed by the handler not the worker
 # Bundle edge Rails instead: gem "rails", github: "rails/rails", branch: "main"
 gem "rails", "~> 7.0.4", ">= 7.0.4.3"
 
@@ -15,30 +16,28 @@ gem "puma", "~> 5.0"
 # Build JSON APIs with ease [https://github.com/rails/jbuilder]
 gem "jbuilder"
 
+# Use Rack CORS for handling Cross-Origin Resource Sharing (CORS), making cross-origin AJAX possible
+gem "rack-cors"
+
+# Reduces boot times through caching; required in config/boot.rb
+gem "bootsnap", require: false
+
 # Use Redis adapter to run Action Cable in production
 gem "redis", "~> 4.0"
 gem 'hiredis'
 
 # DynamoDB for storing registrations
 gem 'aws-sdk-dynamodb'
 
+# SQS for adding data into a queue
+gem 'aws-sdk-sqs'
+
 # Use Kredis to get higher-level data types in Redis [https://github.com/rails/kredis]
 gem "kredis"
 
-# Use Active Model has_secure_password [https://guides.rubyonrails.org/active_model_basics.html#securepassword]
-# gem "bcrypt", "~> 3.1.7"
-
 # Windows does not include zoneinfo files, so bundle the tzinfo-data gem
 gem "tzinfo-data", platforms: %i[ mingw mswin x64_mingw jruby ]
 
-# Reduces boot times through caching; required in config/boot.rb
-gem "bootsnap", require: false
-
-# Use Active Storage variants [https://guides.rubyonrails.org/active_storage_overview.html#transforming-images]
-# gem "image_processing", "~> 1.2"
-
-# Use Rack CORS for handling Cross-Origin Resource Sharing (CORS), making cross-origin AJAX possible
-gem "rack-cors"
 
 group :development, :test do
   # See https://guides.rubyonrails.org/debugging_rails_applications.html#debugging-with-the-debug-gem

Gemfile.lock

@@ -76,6 +76,9 @@ GEM
     aws-sdk-dynamodb (1.84.0)
       aws-sdk-core (~> 3, >= 3.165.0)
       aws-sigv4 (~> 1.1)
+    aws-sdk-sqs (1.55.0)
+      aws-sdk-core (~> 3, >= 3.165.0)
+      aws-sigv4 (~> 1.1)
     aws-sigv4 (1.5.2)
       aws-eventstream (~> 1, >= 1.0.2)
     bootsnap (1.16.0)
@@ -181,6 +184,7 @@ PLATFORMS
 
 DEPENDENCIES
   aws-sdk-dynamodb
+  aws-sdk-sqs
   bootsnap
   debug
   hiredis

app/controllers/metrics_controller.rb

@@ -0,0 +1,16 @@
+require 'securerandom'
+class MetricsController < ApplicationController
+  def index
+    # Get the queue attributes
+    queue_url = $sqs.get_queue_url(queue_name: "registrations.fifo").queue_url
+    response = $sqs.get_queue_attributes({
+                queue_url: queue_url,
+                attribute_names: ["ApproximateNumberOfMessages"]
+              })
+
+    # Get the queue size
+    queue_size = response.attributes["ApproximateNumberOfMessages"].to_i
+
+    render json: { queue_size: queue_size}
+  end
+end

app/controllers/registration_controller.rb

@@ -5,27 +5,95 @@ def create
     competition_id = params[:competition_id]
     event_ids = params[:event_ids]
 
-    unless user_can_register(competitor_id, competition_id)
+    unless validate_request(competitor_id, competition_id)
       return render json: { status: 'User cannot register, wrong format' }, status: :forbidden
     end
 
-    registration = {
-      id: SecureRandom.uuid,
+    id = SecureRandom.uuid
+
+    step_data = {
       competitor_id: competitor_id,
       competition_id: competition_id,
-      registration_data: {
-        event_ids: event_ids
-      }
+      event_ids: event_ids,
+      registration_status: "waiting",
+      step: "Event Registration"
     }
+    queue = Aws::SQS::Queue.new($sqs.get_queue_url(queue_name: "registrations.fifo").queue_url)
 
-    $dynamodb.put_item({
-       table_name: 'Registrations',
-       item: registration
-    })
+    queue.send_message({
+     queue_url: $queue,
+     message_body: step_data.to_json,
+     message_group_id: id,
+     message_deduplication_id: id
+     })
+
+    render json: { status: 'ok', message: "Started Registration Process" }
+  end
+
+  def update
+    competitor_id = params[:competitor_id]
+    competition_id = params[:competition_id]
+    status = params[:status]
+
+    unless validate_request(competitor_id, competition_id, status)
+      return render json: { status: 'User cannot register, wrong format' }, status: :forbidden
+    end
+
+    # Specify the key attributes for the item to be updated
+    key = {
+      'competitor_id' => competitor_id,
+      'competition_id' => competition_id
+    }
+
+    # Set the expression for the update operation
+    update_expression = 'set registration_status = :s'
+    expression_attribute_values = {
+      ':s' => status
+    }
 
-    render json: { status: 'ok' }
+    begin
+      # Update the item in the table
+      $dynamodb.update_item({
+         table_name: "Registrations",
+         key: key,
+         update_expression: update_expression,
+         expression_attribute_values: expression_attribute_values
+       })
+      return render json: { status: 'ok' }
+    rescue Aws::DynamoDB::Errors::ServiceError => e
+      return render json: { status: 'Failed to update registration data' }, status: :internal_server_error
+    end
   end
 
+  def delete
+    competitor_id = params[:competitor_id]
+    competition_id = params[:competition_id]
+
+    unless validate_request(competitor_id, competition_id)
+      return render json: { status: 'User cannot register, wrong format' }, status: :forbidden
+    end
+
+    # Define the key of the item to delete
+    key = {
+      "competition_id" => competition_id,
+      "competitor_id" => competitor_id
+    }
+
+    begin
+      # Call the delete_item method to delete the item from the table
+      $dynamodb.delete_item(
+        table_name: "Registrations",
+        key: key
+      )
+
+      # Render a success response
+      return render json: { status: 'ok' }
+
+    rescue Aws::DynamoDB::Errors::ServiceError => error
+      # Render an error response
+      return render json: { status: "Error deleting item from DynamoDB: #{error.message}" }, status: :internal_server_error
+    end
+  end
   def list
     competition_id = params[:competition_id]
     registrations = get_registrations(competition_id)
@@ -35,11 +103,13 @@ def list
 
   private
 
-  def user_can_register(competitor_id, competition_id)
+  REGISTRATION_STATUS = %w[waiting accepted]
+
+  def validate_request(competitor_id, competition_id, status="waiting")
     # check that competitor ID is in the correct format
     if competitor_id =~ /^\d{4}[a-zA-Z]{4}\d{2}$/
       # check that competition ID is in the correct format
-      if competition_id =~ /^[a-zA-Z]+\d{4}$/
+      if competition_id =~ /^[a-zA-Z]+\d{4}$/ and REGISTRATION_STATUS.include? status
         return true
       end
     end

app/worker/queue_poller.rb

@@ -0,0 +1,37 @@
+require 'json'
+require 'aws-sdk-sqs'
+require_relative 'registration_processor'
+
+class QueuePoller
+  # Wait for 1 second so we can start work on 10 messages at at time
+  # These numbers can be tweaked after load testing
+  WAIT_TIME = 1
+  MAX_MESSAGES = 10
+
+  def self.perform
+    if ENV['LOCALSTACK_ENDPOINT']
+      @sqs ||= Aws::SQS::Client.new(endpoint: ENV['LOCALSTACK_ENDPOINT'])
+    else
+      @sqs ||= Aws::SQS::Client.new
+    end
+
+    queue = @sqs.get_queue_url(queue_name: "registrations.fifo").queue_url
+    poller = Aws::SQS::QueuePoller.new(queue)
+    poller.poll(wait_time_seconds: WAIT_TIME, max_number_of_messages: MAX_MESSAGES) do |messages|
+      messages.each do |msg|
+        # Messages are deleted from the queue when the block returns normally!
+        puts "Received message with ID: #{msg.message_id}"
+        puts "Message body: #{msg.body}"
+        body = JSON.parse msg.body
+        begin
+          RegistrationProcessor.process_message(body)
+        rescue StandardError => e
+          # unexpected error occurred while processing messages,
+          # log it, and skip delete so it can be re-processed later
+          puts "Error #{e} when processing message with ID #{msg}"
+          throw :skip_delete
+        end
+      end
+    end
+  end
+end

app/worker/registration_processor.rb

@@ -0,0 +1,31 @@
+require 'aws-sdk-dynamodb'
+
+class RegistrationProcessor
+  def self.process_message(message)
+    if ENV['LOCALSTACK_ENDPOINT']
+      @dynamodb ||= Aws::DynamoDB::Client.new(endpoint: ENV['LOCALSTACK_ENDPOINT'])
+    else
+      @dynamodb ||= Aws::DynamoDB::Client.new
+    end
+    # implement your message processing logic here
+    puts "Working on Message: #{message}"
+    if message['step'] == "Event Registration"
+      registration = {
+        competitor_id: message['competitor_id'],
+        competition_id: message['competition_id'],
+        event_ids: message['event_ids'],
+        registration_status: "waiting",
+      }
+      save_registration(registration)
+    end
+  end
+
+  private
+
+  def self.save_registration(registration)
+    @dynamodb.put_item({
+       table_name: 'Registrations',
+       item: registration
+     })
+  end
+end

config/application.rb

@@ -11,14 +11,6 @@ class Application < Rails::Application
     # Initialize configuration defaults for originally generated Rails version.
     config.load_defaults 7.0
 
-    # Configuration for the application, engines, and railties goes here.
-    #
-    # These settings can be overridden in specific environments using the files
-    # in config/environments, which are processed later.
-    #
-    # config.time_zone = "Central Time (US & Canada)"
-    # config.eager_load_paths << Rails.root.join("extras")
-
     # Only loads a smaller set of middleware suitable for API only apps.
     # Middleware like session, flash, cookies can be added back manually.
     # Skip views, helpers and assets when generating a new resource.

config/initializers/aws.rb

@@ -1,14 +1,11 @@
 # config/initializers/aws.rb
 
 if Rails.env.production?
-  # We are using IAM Roles to authenticate in prod
-  Aws.config.update({
-    region: ENV["AWS_REGION"],
-  })
   $dynamodb = Aws::DynamoDB::Client.new
+  $sqs = Aws::SQS::Client.new
+  $queue = ENV["QUEUE_URL"]
 else
-  # We are using fake values in dev
-  $dynamodb = Aws::DynamoDB::Client.new(endpoint: ENV['DYNAMODB_ENDPOINT'], region: "my-cool-region-1", credentials: Aws::Credentials.new('my_cool_key', 'my_cool_secret'))
+  # We are using localstack to emulate AWS in dev
+  $dynamodb = Aws::DynamoDB::Client.new(endpoint: ENV['LOCALSTACK_ENDPOINT'])
+  $sqs = Aws::SQS::Client.new(endpoint: ENV['LOCALSTACK_ENDPOINT'])
 end
-
-

config/puma.rb

@@ -24,18 +24,18 @@
 # Specifies the `pidfile` that Puma will use.
 pidfile ENV.fetch("PIDFILE") { "tmp/pids/server.pid" }
 
-# Specifies the number of `workers` to boot in clustered mode.
-# Workers are forked web server processes. If using threads and workers together
-# the concurrency of the application would be max `threads` * `workers`.
+# Specifies the number of `worker` to boot in clustered mode.
+# Workers are forked web server processes. If using threads and worker together
+# the concurrency of the application would be max `threads` * `worker`.
 # Workers do not work on JRuby or Windows (both of which do not support
 # processes).
 #
-# workers ENV.fetch("WEB_CONCURRENCY") { 2 }
+# worker ENV.fetch("WEB_CONCURRENCY") { 2 }
 
-# Use the `preload_app!` method when specifying a `workers` number.
+# Use the `preload_app!` method when specifying a `worker` number.
 # This directive tells Puma to first boot the application and load code
 # before forking the application. This takes advantage of Copy On Write
-# process behavior so workers use less memory.
+# process behavior so worker use less memory.
 #
 # preload_app!
 

config/routes.rb

@@ -1,5 +1,8 @@
 Rails.application.routes.draw do
   get '/healthcheck', to: 'healthcheck#index'
   post '/register', to: 'registration#create'
+  patch '/register', to: 'registration#update'
+  delete '/register', to: 'registration#delete'
   get '/registrations', to: 'registration#list'
+  get '/metrics', to: 'metrics#index'
 end

db/seeds.rb

@@ -1,3 +1,4 @@
+# Create the DynamoDB Tables
 table_name = 'Registrations'
 key_schema = [
   { attribute_name: 'competition_id', key_type: 'HASH' },
@@ -17,3 +18,12 @@
  attribute_definitions: attribute_definitions,
  provisioned_throughput: provisioned_throughput
 })
+
+# Create SQS Queue
+queue_name = 'registrations.fifo'
+$sqs.create_queue({
+                    queue_name: queue_name,
+                    attributes: {
+                      "FifoQueue": "true"
+                    }
+                  })