shoryuken

Author: sorah

Gemfile

@@ -15,6 +15,9 @@ gem 'aws-sdk-chimesdkidentity'
 gem 'aws-sdk-chimesdkmessaging'
 gem 'aws-sdk-ivs'
 
+gem 'shoryuken'
+gem 'aws-sdk-sqs'
+
 gem 'simpacker'
 gem 's3_assets_uploader'
 

Gemfile.lock

@@ -83,6 +83,9 @@ GEM
       aws-sdk-core (~> 3, >= 3.119.0)
       aws-sdk-kms (~> 1)
       aws-sigv4 (~> 1.1)
+    aws-sdk-sqs (1.43.0)
+      aws-sdk-core (~> 3, >= 3.119.0)
+      aws-sigv4 (~> 1.1)
     aws-sdk-transcribestreamingservice (1.31.0)
       aws-sdk-core (~> 3, >= 3.119.0)
       aws-sigv4 (~> 1.1)
@@ -208,6 +211,10 @@ GEM
     sentry-ruby-core (4.7.1)
       concurrent-ruby
       faraday
+    shoryuken (5.2.3)
+      aws-sdk-core (>= 2)
+      concurrent-ruby
+      thor
     simpacker (1.2.0)
       railties (>= 4.2)
     sprockets (4.0.2)
@@ -236,6 +243,7 @@ DEPENDENCIES
   aws-sdk-core
   aws-sdk-ivs
   aws-sdk-s3
+  aws-sdk-sqs
   aws-sdk-transcribestreamingservice
   barnes
   byebug
@@ -250,6 +258,7 @@ DEPENDENCIES
   sass-rails (>= 6)
   sentry-rails
   sentry-ruby
+  shoryuken
   simpacker
 
 BUNDLED WITH

Procfile

@@ -1,2 +1,3 @@
 release: bundle exec rake db:migrate assets:upload
 web: bundle exec puma -p $PORT -C config/puma.rb
+worker: bundle exec shoryuken start -R -C config/shoryuken.yml

app/controllers/api/control/track_cards_controller.rb

@@ -12,7 +12,7 @@ def index
 
   def create
     TrackCard.create!(track_card_params)
-    EmitIvsMetadataJob.perform_later
+    EmitIvsMetadataJob.perform_now
     render(json: {ok: true}.to_json)
   end
 

app/jobs/application_job.rb

@@ -1,2 +1,3 @@
 class ApplicationJob < ActiveJob::Base
+  queue_as ENV["TAKEOUT_SHORYUKEN_QUEUE"] if ENV["TAKEOUT_SHORYUKEN_QUEUE"]
 end

config/initializers/shoryuken.rb

@@ -0,0 +1,39 @@
+module Shoryuken
+  module Middleware
+    module Server
+      class RavenReporter
+        def call(worker_instance, queue, sqs_msg, body)
+          #tags = {}
+          #context = { message: body }
+          Sentry.with_scope do
+            scope.set_transaction_name(body['job_class'])
+            scope.set_tags( job: body['job_class'], queue: queue )
+            begin
+              yield
+            rescue Exception => e
+              Sentry.capture_exception(e)
+              raise
+            end
+          end
+        end
+      end
+    end
+  end
+end
+
+Shoryuken.active_job_queue_name_prefixing = false
+
+if ENV['TAKEOUT_SHORYUKEN_QUEUE']
+  Shoryuken.configure_client do |config|
+    config.sqs_client = Aws::SQS::Client.new(region: ENV.fetch('TAKEOUT_SHORYUKEN_AWS_REGION'), logger: Rails.logger)
+  end
+  Shoryuken.configure_server do |config|
+    config.server_middleware do |chain|
+      chain.add Shoryuken::Middleware::Server::RavenReporter
+    end
+
+    config.sqs_client = Aws::SQS::Client.new(region: ENV.fetch('TAEKOUT_SHORUKEN_AWS_REGION'), logger: Rails.logger)
+    # see also config/shoryuken.yml
+    config.sqs_client_receive_message_opts = { wait_time_seconds: 20 } 
+  end
+end

config/shoryuken.yml

@@ -0,0 +1,4 @@
+concurrency: <%= ENV.fetch('RAILS_MAX_THREADS', 10) %>
+# ActiveJob prefixes queue name, the default delimiter is `_`
+queues:
+  - [<%= ENV.fetch('TAKEOUT_SHORYUKEN_QUEUE') %>, 1]

tf/iam_heroku-takeout-prd.tf

@@ -82,6 +82,27 @@ data "aws_iam_policy_document" "heroku-takeout-prd" {
       "arn:aws:ivs:us-west-2:005216166247:channel/oTssPyKzhjoS",
     ]
   }
+
+  statement {
+    effect = "Allow"
+    actions = [
+      "sqs:ChangeMessageVisibility",
+      "sqs:ChangeMessageVisibilityBatch",
+      "sqs:DeleteMessage",
+      "sqs:DeleteMessageBatch",
+      "sqs:GetQueueAttributes",
+      "sqs:GetQueueUrl",
+      "sqs:ReceiveMessage",
+      "sqs:SendMessage",
+      "sqs:SendMessageBatch",
+      "sqs:ListQueues",
+    ]
+    resources = [
+      aws_sqs_queue.activejob-prd.arn,
+      aws_sqs_queue.activejob-dlq-prd.arn,
+    ]
+  }
+
 }
 
 resource "aws_iam_user_policy" "heroku-takeout-prd" {

tf/sqs.tf

@@ -0,0 +1,23 @@
+resource "aws_sqs_queue" "activejob-prd" {
+  provider = aws.use1
+  name     = "takeout-app-activejob-prd"
+
+  redrive_policy = jsonencode({
+    deadLetterTargetArn = aws_sqs_queue.activejob-dlq-prd.arn
+    maxReceiveCount     = 10
+  })
+
+  tags = {
+    Environment = "production"
+  }
+}
+
+
+resource "aws_sqs_queue" "activejob-dlq-prd" {
+  provider = aws.use1
+  name     = "takeout-app-activejob-dlq-prd"
+
+  tags = {
+    Environment = "production"
+  }
+}