[ST-2002] Totems Timeout

lib/lita/handlers/totems.rb

@@ -2,6 +2,10 @@
 require 'chronic_duration'
 require 'redis-semaphore'
 require "lita/handlers/stats"
+require "lita/handlers/workers"
+require 'sidekiq'
+require 'sidekiq/api'
+
 module Lita
   module Handlers
     class Totems < Handler
@@ -27,6 +31,10 @@ class Totems < Handler
         etl_extract:"#squad-data-platform"
       }
 
+      def self.DemoEnvironments
+        @@DemoEnvironments
+      end
+
       def self.route_regex(action_capture_group)
         %r{
         ^totems?\s+
@@ -313,14 +321,8 @@ def take_totem(response, totem, user_id, timeout)
         redis.sadd("user/#{user_id}/totems", totem)
         redis.hset("totem/#{totem}/waiting_since", user_id, Time.now.to_i)
         if @@DemoEnvironments.include? totem
-          # Create async job
-          after(timeout*3600) do |timer|
-            # Check that the user is the current owner of the totem
-            current_owner = redis.get("totem/#{totem}/owning_user_id")
-            if user_id == current_owner
-              yield_totem(response.match_data[:totem], user_id, response, true)
-            end
-          end
+          timeout_job = Timeout.perform_in(timeout, user_id, totem)
+          redis.hset("totem/#{totem}/timeout_jobs", user_id, timeout_job.jid)
         end
       end
 
@@ -330,7 +332,7 @@ def get_user_by_id(user_id)
         Lita::User.find_by_id(user_id)
       end
 
-      def yield_totem(totem, user_id, response, timeout=false)
+      def yield_totem(totem, user_id, response)
         waiting_since_hash = redis.hgetall("totem/#{totem}/waiting_since")
         Stats.capture_holding_time(totem, waiting_since_hash[user_id])    
 
@@ -340,27 +342,26 @@ def yield_totem(totem, user_id, response, timeout=false)
         redis.hdel("totem/#{totem}/timeout", user_id)
         redis.srem("user/#{user_id}/totems/reminder", totem) if redis.smembers("user/#{user_id}/totems/reminder").include?(totem)
         next_user_id = redis.lpop("totem/#{totem}/list")
-        # TODO: Remove async job
-        # TODO: Find a way to identify pending jobs so we can cancel them instead of letting them finish and then checking 
+        timeout_jobs = redis.hgetall("totem/#{totem}/timeout_jobs")
+        current_timeout_job = timeout_jobs[user_id]
+        if current_timeout_job
+          Sidekiq::Queue.new
+          queue.each do |job|
+            user_id
+            job.delete if job.jid == current_timeout_job
+          end
+        end
         if next_user_id
           timeout_hash = redis.hgetall("totem/#{totem}/timeout")
           Stats.capture_waiting_time(totem, waiting_since_hash[next_user_id])
           take_totem(response, totem, next_user_id, timeout_hash[next_user_id].to_i)
           next_user = Lita::User.find_by_id(next_user_id)
           queue_size = redis.llen("totem/#{totem}/list")
           robot.send_messages(Lita::Source.new(user: next_user), %{You are now in possession of totem "#{totem}", yielded by #{response.user.name}. There are #{queue_size} people in line after you.})
-          if timeout
-            robot.send_messages(Lita::Source.new(user: get_user_by_id(user_id)), %{Your totem "#{totem}", expired and has been given to #{next_user.name}.})
-          else
-            response.reply "You have yielded the totem to #{next_user.name}."
-          end
+          response.reply "You have yielded the totem to #{next_user.name}."
         else
           redis.del("totem/#{totem}/owning_user_id")
-          if timeout
-            robot.send_messages(Lita::Source.new(user: get_user_by_id(user_id)), %{Your totem "#{totem}" has expired.})
-          else
-            response.reply %{You have yielded the "#{totem}" totem.}
-          end
+          response.reply %{You have yielded the "#{totem}" totem.}
         end
       end
 
@@ -379,6 +380,17 @@ def setReminderToTotemOwner(current_owner_id, response, totem)
       end
     end
 
+    Thread.new { Timer.new(interval: 5, recurring: true) do |timer|
+      robot = Robot.new
+      Totems.DemoEnvironments.each do |totem|
+        pending_messages = Lita.redis.hgetall("totem/#{totem}/timeout_messages")
+        pending_messages.each do |user_id, message|
+          robot.send_messages(Lita::Source.new(user: Lita::User.find_by_id(user_id)), message)
+          Lita.redis.hdel("totem/#{totem}/timeout_messages", user_id)
+        end
+      end
+    end.start }
+
     Lita.register_handler(Totems)
   end
 end

lib/lita/handlers/workers.rb

@@ -0,0 +1,62 @@
+require 'sidekiq'
+require 'lita'
+require "lita/handlers/stats"
+
+module Lita
+  module Handlers
+    class Timeout
+      include Sidekiq::Job
+
+      def initialize
+        Sidekiq.redis do |connection| 
+          @redis_namespace = Redis::Namespace.new("lita:handlers:#{Totems.namespace}", redis: connection)
+        end
+      end
+
+      def take_totem(totem, user_id, timeout)
+        @redis_namespace.set("totem/#{totem}/owning_user_id", user_id)
+        @redis_namespace.sadd("user/#{user_id}/totems", totem)
+        @redis_namespace.hset("totem/#{totem}/waiting_since", user_id, Time.now.to_i)
+        timeout_job = Timeout.perform_in(timeout, user_id, totem)
+        @redis_namespace.hset("totem/#{totem}/timeout_jobs", user_id, timeout_job.jid)
+      end
+
+      def yield_totem(totem, user_id)
+        waiting_since_hash = @redis_namespace.hgetall("totem/#{totem}/waiting_since")
+        Stats.capture_holding_time(totem, waiting_since_hash[user_id])    
+
+        @redis_namespace.srem("user/#{user_id}/totems", totem)
+        @redis_namespace.hdel("totem/#{totem}/waiting_since", user_id)
+        @redis_namespace.hdel("totem/#{totem}/message", user_id)
+        @redis_namespace.hdel("totem/#{totem}/timeout", user_id)
+        @redis_namespace.srem("user/#{user_id}/totems/reminder", totem) if @redis_namespace.smembers("user/#{user_id}/totems/reminder").include?(totem)
+        next_user_id = @redis_namespace.lpop("totem/#{totem}/list")
+        if next_user_id
+          timeout_hash = @redis_namespace.hgetall("totem/#{totem}/timeout")
+          Stats.capture_waiting_time(totem, waiting_since_hash[next_user_id])
+          take_totem(response, totem, next_user_id, timeout_hash[next_user_id].to_i)
+          next_user = Lita::User.find_by_id(next_user_id)
+          queue_size = @redis_namespace.llen("totem/#{totem}/list")
+          @redis_namespace.hset("totem/#{totem}/timeout_messages", next_user_id, %{You are now in possession of totem "#{totem}", yielded by #{response.user.name}. There are #{queue_size} people in line after you.})
+          @redis_namespace.hset("totem/#{totem}/timeout_messages", user_id, %{Your totem "#{totem}", expired and has been given to #{next_user.name}.})
+        else
+          @redis_namespace.del("totem/#{totem}/owning_user_id")
+          @redis_namespace.hset("totem/#{totem}/timeout_messages", user_id, %{Your totem "#{totem}" has expired.})
+        end
+      end
+
+      def get_user_by_id(user_id)
+        # This query by user_id is needed because if we take the user
+        # from the response, then it will always be the user that first timed out
+        Lita::User.find_by_id(user_id)
+      end
+
+      def notify()
+      end
+
+      def perform(totem, user_id)
+        yield_totem(totem, user_id)
+      end
+    end
+  end
+end