From 0034cbda8ae833962a63e0a77b7f5c84e2d8212d Mon Sep 17 00:00:00 2001
From: Martin Brennan <mjrbrennan@gmail.com>
Date: Tue, 19 Jan 2021 13:30:58 +1000
Subject: [PATCH] DEV: Change Topic Timer from enqueue_at scheduled jobs to
 incrementally executed jobs (#11698)

Moves the topic timer jobs from being scheduled ahead of time with enqueue_at to a 5 minute scheduled run like bookmark reminders, in a new job called Jobs::EnqueueTopicTimers. Backwards compatibility is maintained by checking if an existing topic timer job is enqueued in sidekiq for the timer, and if it is not running it inside the new job.

The functionality to close/open a topic if it is in the opposite state still remains in the after_save block of TopicTimer, with further commentary, which is used for Open/Close Temporarily.

This also removes the ensure_consistency! functionality of topic timers as it is no longer needed; the new job will always pick up the timers because they are not stored in a fragile state of sidekiq.
---
 app/jobs/regular/bump_topic.rb                |  15 +-
 app/jobs/regular/clear_slow_mode.rb           |  18 +--
 app/jobs/regular/close_topic.rb               |  12 +-
 app/jobs/regular/delete_replies.rb            |  13 +-
 app/jobs/regular/delete_topic.rb              |  20 +--
 app/jobs/regular/open_topic.rb                |  13 +-
 app/jobs/regular/publish_topic_to_category.rb |  10 +-
 app/jobs/regular/topic_timer_base.rb          |  20 +++
 app/jobs/scheduled/periodical_updates.rb      |   3 -
 app/jobs/scheduled/topic_timer_enqueuer.rb    |  25 +++
 app/models/topic.rb                           |   2 +
 app/models/topic_timer.rb                     | 134 ++++++++++------
 spec/integration/topic_auto_close_spec.rb     |  12 --
 spec/jobs/publish_topic_to_category_spec.rb   |   4 +-
 spec/jobs/topic_timer_enqueuer_spec.rb        |  50 ++++++
 spec/models/post_action_spec.rb               |   4 +-
 spec/models/topic_spec.rb                     |   2 +-
 spec/models/topic_timer_spec.rb               | 149 +++++++-----------
 spec/support/sidekiq_helpers.rb               |  16 +-
 19 files changed, 278 insertions(+), 244 deletions(-)
 create mode 100644 app/jobs/regular/topic_timer_base.rb
 create mode 100644 app/jobs/scheduled/topic_timer_enqueuer.rb
 create mode 100644 spec/jobs/topic_timer_enqueuer_spec.rb

diff --git a/app/jobs/regular/bump_topic.rb b/app/jobs/regular/bump_topic.rb
index 5a60883e8f1..a8cd9fa35b4 100644
--- a/app/jobs/regular/bump_topic.rb
+++ b/app/jobs/regular/bump_topic.rb
@@ -1,22 +1,13 @@
 # frozen_string_literal: true
 
 module Jobs
-  class BumpTopic < ::Jobs::Base
-
-    def execute(args)
-      topic_timer = TopicTimer.find_by(id: args[:topic_timer_id])
-
-      topic = topic_timer&.topic
-
-      if topic_timer.blank? || topic.blank? || topic_timer.execute_at > Time.zone.now
-        return
-      end
-
+  class BumpTopic < ::Jobs::TopicTimerBase
+    def execute_timer_action(topic_timer, topic)
       if Guardian.new(topic_timer.user).can_create_post_on_topic?(topic)
         topic.add_small_action(Discourse.system_user, "autobumped", nil, bump: true)
       end
+
       topic_timer.trash!(Discourse.system_user)
     end
-
   end
 end
diff --git a/app/jobs/regular/clear_slow_mode.rb b/app/jobs/regular/clear_slow_mode.rb
index 0e7f0cfcde9..361cce87461 100644
--- a/app/jobs/regular/clear_slow_mode.rb
+++ b/app/jobs/regular/clear_slow_mode.rb
@@ -1,22 +1,8 @@
 # frozen_string_literal: true
 
 module Jobs
-  class ClearSlowMode < ::Jobs::Base
-
-    def execute(args)
-      topic_timer = TopicTimer.find_by(id: args[:topic_timer_id])
-
-      if topic_timer.nil? || topic_timer.execute_at > Time.zone.now
-        return
-      end
-
-      topic = topic_timer&.topic
-
-      if topic.nil?
-        topic_timer.destroy!
-        return
-      end
-
+  class ClearSlowMode < ::Jobs::TopicTimerBase
+    def execute_timer_action(topic_timer, topic)
       topic.update!(slow_mode_seconds: 0)
       topic_timer.trash!(Discourse.system_user)
     end
diff --git a/app/jobs/regular/close_topic.rb b/app/jobs/regular/close_topic.rb
index 96364d5841c..044ee64827b 100644
--- a/app/jobs/regular/close_topic.rb
+++ b/app/jobs/regular/close_topic.rb
@@ -1,16 +1,12 @@
 # frozen_string_literal: true
 
 module Jobs
-  class CloseTopic < ::Jobs::Base
-    def execute(args)
-      topic_timer = TopicTimer.find_by(id: args[:topic_timer_id])
-      return if !topic_timer&.runnable?
-
-      topic = topic_timer.topic
+  class CloseTopic < ::Jobs::TopicTimerBase
+    def execute_timer_action(topic_timer, topic)
+      silent = @args[:silent]
       user = topic_timer.user
-      silent = args[:silent]
 
-      if topic.blank? || topic.closed?
+      if topic.closed?
         topic_timer.destroy!
         return
       end
diff --git a/app/jobs/regular/delete_replies.rb b/app/jobs/regular/delete_replies.rb
index 0e5605a8366..895c1534093 100644
--- a/app/jobs/regular/delete_replies.rb
+++ b/app/jobs/regular/delete_replies.rb
@@ -1,17 +1,8 @@
 # frozen_string_literal: true
 
 module Jobs
-  class DeleteReplies < ::Jobs::Base
-
-    def execute(args)
-      topic_timer = TopicTimer.find_by(id: args[:topic_timer_id])
-
-      topic = topic_timer&.topic
-
-      if topic_timer.blank? || topic.blank? || topic_timer.execute_at > Time.zone.now
-        return
-      end
-
+  class DeleteReplies < ::Jobs::TopicTimerBase
+    def execute_timer_action(topic_timer, topic)
       unless Guardian.new(topic_timer.user).is_staff?
         topic_timer.trash!(Discourse.system_user)
         return
diff --git a/app/jobs/regular/delete_topic.rb b/app/jobs/regular/delete_topic.rb
index 829a8331197..af109d8b101 100644
--- a/app/jobs/regular/delete_topic.rb
+++ b/app/jobs/regular/delete_topic.rb
@@ -1,23 +1,17 @@
 # frozen_string_literal: true
 
 module Jobs
-  class DeleteTopic < ::Jobs::Base
-
-    def execute(args)
-      topic_timer = TopicTimer.find_by(id: args[:topic_timer_id])
-
-      topic = topic_timer&.topic
-
-      if topic_timer.blank? || topic.blank? || topic_timer.execute_at > Time.zone.now
-        return
-      end
-
+  class DeleteTopic < ::Jobs::TopicTimerBase
+    def execute_timer_action(topic_timer, topic)
       if Guardian.new(topic_timer.user).can_delete?(topic)
         first_post = topic.ordered_posts.first
-        PostDestroyer.new(topic_timer.user, first_post, context: I18n.t("topic_statuses.auto_deleted_by_timer")).destroy
+
+        PostDestroyer.new(
+          topic_timer.user, first_post, context: I18n.t("topic_statuses.auto_deleted_by_timer")
+        ).destroy
+
         topic_timer.trash!(Discourse.system_user)
       end
     end
-
   end
 end
diff --git a/app/jobs/regular/open_topic.rb b/app/jobs/regular/open_topic.rb
index 36246cb6531..85f2b96afc6 100644
--- a/app/jobs/regular/open_topic.rb
+++ b/app/jobs/regular/open_topic.rb
@@ -1,19 +1,10 @@
 # frozen_string_literal: true
 
 module Jobs
-  class OpenTopic < ::Jobs::Base
-    def execute(args)
-      topic_timer = TopicTimer.find_by(id: args[:topic_timer_id])
-      return if !topic_timer&.runnable?
-
-      topic = topic_timer.topic
+  class OpenTopic < ::Jobs::TopicTimerBase
+    def execute_timer_action(topic_timer, topic)
       user = topic_timer.user
 
-      if topic.blank?
-        topic_timer.destroy!
-        return
-      end
-
       if !Guardian.new(user).can_open_topic?(topic) || topic.open?
         topic_timer.destroy!
         topic.reload
diff --git a/app/jobs/regular/publish_topic_to_category.rb b/app/jobs/regular/publish_topic_to_category.rb
index f3c41148388..b42079dbf64 100644
--- a/app/jobs/regular/publish_topic_to_category.rb
+++ b/app/jobs/regular/publish_topic_to_category.rb
@@ -1,14 +1,8 @@
 # frozen_string_literal: true
 
 module Jobs
-  class PublishTopicToCategory < ::Jobs::Base
-    def execute(args)
-      topic_timer = TopicTimer.find_by(id: args[:topic_timer_id])
-      return if topic_timer.blank?
-
-      topic = topic_timer.topic
-      return if topic.blank?
-
+  class PublishTopicToCategory < ::Jobs::TopicTimerBase
+    def execute_timer_action(topic_timer, topic)
       return unless Guardian.new(topic_timer.user).can_see?(topic)
 
       TopicTimer.transaction do
diff --git a/app/jobs/regular/topic_timer_base.rb b/app/jobs/regular/topic_timer_base.rb
new file mode 100644
index 00000000000..35b582a97b4
--- /dev/null
+++ b/app/jobs/regular/topic_timer_base.rb
@@ -0,0 +1,20 @@
+# frozen_string_literal: true
+
+module Jobs
+  class TopicTimerBase < ::Jobs::Base
+    def execute(args)
+      @args = args
+
+      topic_timer = TopicTimer.find_by(id: args[:topic_timer_id])
+      return if !topic_timer&.runnable?
+
+      topic = topic_timer.topic
+      if topic.blank?
+        topic_timer.destroy!
+        return
+      end
+
+      execute_timer_action(topic_timer, topic)
+    end
+  end
+end
diff --git a/app/jobs/scheduled/periodical_updates.rb b/app/jobs/scheduled/periodical_updates.rb
index 2ae53c951d2..5c5da8bac6b 100644
--- a/app/jobs/scheduled/periodical_updates.rb
+++ b/app/jobs/scheduled/periodical_updates.rb
@@ -24,9 +24,6 @@ module Jobs
       args[:max_topic_length] = 500 unless self.class.should_update_long_topics?
       ScoreCalculator.new.calculate(args)
 
-      # Re-run stuff that we missed
-      TopicTimer.ensure_consistency!
-
       # Forces rebake of old posts where needed, as long as no system avatars need updating
       if !SiteSetting.automatically_download_gravatars || !UserAvatar.where("last_gravatar_download_attempt IS NULL").limit(1).first
         problems = Post.rebake_old(SiteSetting.rebake_old_posts_count, priority: :ultra_low)
diff --git a/app/jobs/scheduled/topic_timer_enqueuer.rb b/app/jobs/scheduled/topic_timer_enqueuer.rb
new file mode 100644
index 00000000000..7fa2889b1f3
--- /dev/null
+++ b/app/jobs/scheduled/topic_timer_enqueuer.rb
@@ -0,0 +1,25 @@
+# frozen_string_literal: true
+
+module Jobs
+
+  # Runs periodically to look through topic timers that are ready to execute,
+  # and enqueues their related jobs.
+  #
+  # Any leftovers will be caught in the next run, because execute_at will
+  # be < now, and topic timers that have run are deleted on completion or
+  # otherwise have their execute_at time modified.
+  class TopicTimerEnqueuer < ::Jobs::Scheduled
+    every 1.minute
+
+    def execute(_args = nil)
+      timers = TopicTimer.pending_timers
+
+      timers.find_each do |timer|
+
+        # the typed job may not enqueue if it has already
+        # been scheduled with enqueue_at
+        timer.enqueue_typed_job
+      end
+    end
+  end
+end
diff --git a/app/models/topic.rb b/app/models/topic.rb
index 46a522cbd61..a6cc367baf7 100644
--- a/app/models/topic.rb
+++ b/app/models/topic.rb
@@ -1357,6 +1357,8 @@ class Topic < ActiveRecord::Base
       end
 
       if self.persisted?
+        # See TopicTimer.after_save for additional context; the topic
+        # status may be changed by saving.
         topic_timer.save!
       else
         self.topic_timers << topic_timer
diff --git a/app/models/topic_timer.rb b/app/models/topic_timer.rb
index 053742e1723..613ea2a532e 100644
--- a/app/models/topic_timer.rb
+++ b/app/models/topic_timer.rb
@@ -18,6 +18,9 @@ class TopicTimer < ActiveRecord::Base
   validate :executed_at_in_future?
 
   scope :scheduled_bump_topics, -> { where(status_type: TopicTimer.types[:bump], deleted_at: nil).pluck(:topic_id) }
+  scope :pending_timers, ->(before_time = Time.now.utc) do
+    where("execute_at <= :before_time AND deleted_at IS NULL", before_time: before_time)
+  end
 
   before_save do
     self.created_at ||= Time.zone.now if execute_at
@@ -27,21 +30,68 @@ class TopicTimer < ActiveRecord::Base
        !attribute_in_database(:execute_at).nil?) ||
        will_save_change_to_user_id?
 
-      # private implementation detail have to use send
+      # TODO(martin - 2021-05-01) - Remove this backwards compatability for outstanding
+      # jobs once they have all been run and after Jobs::TopicTimerEnqueuer is in place
       self.send("cancel_auto_#{self.class.types[status_type]}_job")
     end
   end
 
+  # These actions are in place to make sure the topic is in the correct
+  # state at the point in time where the timer is saved. It does not
+  # guarantee that the topic will be in the correct state when the timer
+  # job is executed, but each timer job handles deleted topics etc. gracefully.
+  #
+  # This is also important for the Open Temporarily and Close Temporarily timers,
+  # which change the topic's status straight away and set a timer to do the
+  # opposite action in the future.
   after_save do
     if (saved_change_to_execute_at? || saved_change_to_user_id?)
-      now = Time.zone.now
-      time = execute_at < now ? now : execute_at
-
-      # private implementation detail have to use send
-      self.send("schedule_auto_#{self.class.types[status_type]}_job", time)
+      if status_type == TopicTimer.types[:silent_close] || status_type == TopicTimer.types[:close]
+        topic.update_status('closed', false, user) if topic.closed?
+      end
+      if status_type == TopicTimer.types[:open]
+        topic.update_status('closed', true, user) if topic.open?
+      end
     end
   end
 
+  def status_type_name
+    self.class.types[status_type]
+  end
+
+  def enqueue_typed_job(time: nil)
+    return if typed_job_scheduled?
+    self.send("schedule_auto_#{status_type_name}_job")
+  end
+
+  # TODO(martin - 2021-05-01) - Remove this backwards compatability for outstanding
+  # jobs once they have all been run and after Jobs::TopicTimerEnqueuer is in place
+  def typed_job_scheduled?
+    scheduled = Jobs.scheduled_for(
+      TopicTimer.type_job_map[status_type_name], topic_timer_id: id
+    ).any?
+
+    if [:close, :silent_close, :open].include?(status_type_name)
+      return scheduled || Jobs.scheduled_for(:toggle_topic_closed, topic_timer_id: id).any?
+    end
+
+    scheduled
+  end
+
+  def self.type_job_map
+    {
+      close: :close_topic,
+      open: :open_topic,
+      publish_to_category: :publish_topic_to_category,
+      delete: :delete_topic,
+      reminder: :topic_reminder,
+      bump: :bump_topic,
+      delete_replies: :delete_replies,
+      silent_close: :close_topic,
+      clear_slow_mode: :clear_slow_mode
+    }
+  end
+
   def self.types
     @types ||= Enum.new(
       close: 1,
@@ -64,18 +114,6 @@ class TopicTimer < ActiveRecord::Base
     @_private_types ||= types.only(:reminder, :clear_slow_mode)
   end
 
-  def self.ensure_consistency!
-    TopicTimer.where("topic_timers.execute_at < ?", Time.zone.now)
-      .find_each do |topic_timer|
-
-      # private implementation detail scoped to class
-      topic_timer.send(
-        "schedule_auto_#{self.types[topic_timer.status_type]}_job",
-        topic_timer.execute_at
-      )
-    end
-  end
-
   def public_type?
     !!self.class.public_types[self.status_type]
   end
@@ -100,6 +138,10 @@ class TopicTimer < ActiveRecord::Base
     ))
   end
 
+  def publishing_to_category?
+    self.status_type.to_i == TopicTimer.types[:publish_to_category]
+  end
+
   # TODO(martin - 2021-05-01) - Remove cancels for toggle_topic_closed once topic timer revamp completed.
   def cancel_auto_close_job
     Jobs.cancel_scheduled_job(:toggle_topic_closed, topic_timer_id: id)
@@ -119,73 +161,63 @@ class TopicTimer < ActiveRecord::Base
   end
 
   def cancel_auto_publish_to_category_job
-    Jobs.cancel_scheduled_job(:publish_topic_to_category, topic_timer_id: id)
+    Jobs.cancel_scheduled_job(TopicTimer.type_job_map[:publish_to_category], topic_timer_id: id)
   end
 
   def cancel_auto_delete_job
-    Jobs.cancel_scheduled_job(:delete_topic, topic_timer_id: id)
+    Jobs.cancel_scheduled_job(TopicTimer.type_job_map[:delete], topic_timer_id: id)
   end
 
   def cancel_auto_reminder_job
-    Jobs.cancel_scheduled_job(:topic_reminder, topic_timer_id: id)
+    Jobs.cancel_scheduled_job(TopicTimer.type_job_map[:reminder], topic_timer_id: id)
   end
 
   def cancel_auto_bump_job
-    Jobs.cancel_scheduled_job(:bump_topic, topic_timer_id: id)
+    Jobs.cancel_scheduled_job(TopicTimer.type_job_map[:bump], topic_timer_id: id)
   end
 
   def cancel_auto_delete_replies_job
-    Jobs.cancel_scheduled_job(:delete_replies, topic_timer_id: id)
+    Jobs.cancel_scheduled_job(TopicTimer.type_job_map[:delete_replies], topic_timer_id: id)
   end
 
   def cancel_auto_clear_slow_mode_job
-    Jobs.cancel_scheduled_job(:clear_slow_mode, topic_timer_id: id)
+    Jobs.cancel_scheduled_job(TopicTimer.type_job_map[:clear_slow_mode], topic_timer_id: id)
   end
 
-  def schedule_auto_delete_replies_job(time)
-    Jobs.enqueue_at(time, :delete_replies, topic_timer_id: id)
+  def schedule_auto_delete_replies_job
+    Jobs.enqueue(TopicTimer.type_job_map[:delete_replies], topic_timer_id: id)
   end
 
-  def schedule_auto_bump_job(time)
-    Jobs.enqueue_at(time, :bump_topic, topic_timer_id: id)
+  def schedule_auto_bump_job
+    Jobs.enqueue(TopicTimer.type_job_map[:bump], topic_timer_id: id)
   end
 
-  def schedule_auto_open_job(time)
-    topic.update_status('closed', true, user) if topic && !topic.closed
-
-    Jobs.enqueue_at(time, :open_topic, topic_timer_id: id)
+  def schedule_auto_open_job
+    Jobs.enqueue(TopicTimer.type_job_map[:open], topic_timer_id: id)
   end
 
-  def schedule_auto_close_job(time)
-    topic.update_status('closed', false, user) if topic&.closed
-
-    Jobs.enqueue_at(time, :close_topic, topic_timer_id: id)
+  def schedule_auto_close_job
+    Jobs.enqueue(TopicTimer.type_job_map[:close], topic_timer_id: id)
   end
 
-  def schedule_auto_silent_close_job(time)
-    topic.update_status('closed', false, user) if topic&.closed
-
-    Jobs.enqueue_at(time, :close_topic, topic_timer_id: id, silent: true)
+  def schedule_auto_silent_close_job
+    Jobs.enqueue(TopicTimer.type_job_map[:close], topic_timer_id: id, silent: true)
   end
 
-  def schedule_auto_publish_to_category_job(time)
-    Jobs.enqueue_at(time, :publish_topic_to_category, topic_timer_id: id)
+  def schedule_auto_publish_to_category_job
+    Jobs.enqueue(TopicTimer.type_job_map[:publish_to_category], topic_timer_id: id)
   end
 
-  def publishing_to_category?
-    self.status_type.to_i == TopicTimer.types[:publish_to_category]
+  def schedule_auto_delete_job
+    Jobs.enqueue(TopicTimer.type_job_map[:delete], topic_timer_id: id)
   end
 
-  def schedule_auto_delete_job(time)
-    Jobs.enqueue_at(time, :delete_topic, topic_timer_id: id)
-  end
-
-  def schedule_auto_reminder_job(time)
+  def schedule_auto_reminder_job
     # noop, TODO(martin 2021-03-11): Remove this after timers migrated and outstanding jobs cancelled
   end
 
-  def schedule_auto_clear_slow_mode_job(time)
-    Jobs.enqueue_at(time, :clear_slow_mode, topic_timer_id: id)
+  def schedule_auto_clear_slow_mode_job
+    Jobs.enqueue(TopicTimer.type_job_map[:clear_slow_mode], topic_timer_id: id)
   end
 end
 
diff --git a/spec/integration/topic_auto_close_spec.rb b/spec/integration/topic_auto_close_spec.rb
index 538259a988b..170a3afc11c 100644
--- a/spec/integration/topic_auto_close_spec.rb
+++ b/spec/integration/topic_auto_close_spec.rb
@@ -42,10 +42,6 @@ describe Topic do
 
           expect(topic_status_update.topic).to eq(topic)
           expect(topic.public_topic_timer.execute_at).to eq_time(2.hours.from_now)
-
-          args = job_klass.jobs.last['args'].first
-
-          expect(args["topic_timer_id"]).to eq(topic.public_topic_timer.id)
         end
 
         context 'topic was created by staff user' do
@@ -60,10 +56,6 @@ describe Topic do
             expect(topic_status_update.topic).to eq(staff_topic)
             expect(topic_status_update.execute_at).to eq_time(2.hours.from_now)
             expect(topic_status_update.user).to eq(Discourse.system_user)
-
-            args = job_klass.jobs.last['args'].first
-
-            expect(args["topic_timer_id"]).to eq(topic_status_update.id)
           end
 
           context 'topic is closed manually' do
@@ -90,10 +82,6 @@ describe Topic do
             expect(topic_status_update.topic).to eq(regular_user_topic)
             expect(topic_status_update.execute_at).to eq_time(2.hours.from_now)
             expect(topic_status_update.user).to eq(Discourse.system_user)
-
-            args = job_klass.jobs.last['args'].first
-
-            expect(args["topic_timer_id"]).to eq(topic_status_update.id)
           end
         end
       end
diff --git a/spec/jobs/publish_topic_to_category_spec.rb b/spec/jobs/publish_topic_to_category_spec.rb
index 7ec605e6e64..585766f8b1a 100644
--- a/spec/jobs/publish_topic_to_category_spec.rb
+++ b/spec/jobs/publish_topic_to_category_spec.rb
@@ -12,7 +12,9 @@ RSpec.describe Jobs::PublishTopicToCategory do
     Fabricate(:topic_timer,
       status_type: TopicTimer.types[:publish_to_category],
       category_id: another_category.id,
-      topic: topic
+      topic: topic,
+      execute_at: 1.minute.ago,
+      created_at: 5.minutes.ago
     )
 
     Fabricate(:post, topic: topic)
diff --git a/spec/jobs/topic_timer_enqueuer_spec.rb b/spec/jobs/topic_timer_enqueuer_spec.rb
new file mode 100644
index 00000000000..b9f21ed12c8
--- /dev/null
+++ b/spec/jobs/topic_timer_enqueuer_spec.rb
@@ -0,0 +1,50 @@
+# frozen_string_literal: true
+
+require 'rails_helper'
+
+RSpec.describe Jobs::TopicTimerEnqueuer do
+  subject { described_class.new }
+
+  fab!(:timer1) do
+    Fabricate(:topic_timer, execute_at: 1.minute.ago, created_at: 1.hour.ago, status_type: TopicTimer.types[:close])
+  end
+  fab!(:timer2) do
+    Fabricate(:topic_timer, execute_at: 1.minute.ago, created_at: 1.hour.ago, status_type: TopicTimer.types[:open])
+  end
+  fab!(:future_timer) do
+    Fabricate(:topic_timer, execute_at: 1.hours.from_now, created_at: 1.hour.ago, status_type: TopicTimer.types[:close])
+  end
+  fab!(:deleted_timer) do
+    Fabricate(:topic_timer, execute_at: 1.minute.ago, created_at: 1.hour.ago, status_type: TopicTimer.types[:close])
+  end
+
+  before do
+    deleted_timer.trash!
+  end
+
+  it "does not enqueue deleted timers" do
+    expect_not_enqueued_with(job: :close_topic, args: { topic_timer_id: deleted_timer.id })
+    subject.execute
+    expect(deleted_timer.topic.reload.closed?).to eq(false)
+  end
+
+  it "does not enqueue future timers" do
+    expect_not_enqueued_with(job: :close_topic, args: { topic_timer_id: future_timer.id })
+    subject.execute
+    expect(future_timer.topic.reload.closed?).to eq(false)
+  end
+
+  it "enqueues the related job" do
+    expect_not_enqueued_with(job: :close_topic, args: { topic_timer_id: deleted_timer.id })
+    expect_not_enqueued_with(job: :close_topic, args: { topic_timer_id: future_timer.id })
+    subject.execute
+    expect_job_enqueued(job: :close_topic, args: { topic_timer_id: timer1.id })
+    expect_job_enqueued(job: :open_topic, args: { topic_timer_id: timer2.id })
+  end
+
+  it "does not re-enqueue a job that has already been scheduled ahead of time in sidekiq (legacy topic timers)" do
+    expect_not_enqueued_with(job: :close_topic, args: { topic_timer_id: timer1.id })
+    Jobs.enqueue_at(1.hours.from_now, :close_topic, topic_timer_id: timer1.id)
+    subject.execute
+  end
+end
diff --git a/spec/models/post_action_spec.rb b/spec/models/post_action_spec.rb
index 3f7e80d50a3..26df835d891 100644
--- a/spec/models/post_action_spec.rb
+++ b/spec/models/post_action_spec.rb
@@ -764,9 +764,7 @@ describe PostAction do
 
         freeze_time timer.execute_at
 
-        expect_enqueued_with(job: :open_topic, args: { topic_timer_id: timer.id }, at: Time.zone.now + 1.hour) do
-          Jobs::OpenTopic.new.execute(topic_timer_id: timer.id)
-        end
+        Jobs::OpenTopic.new.execute(topic_timer_id: timer.id)
 
         expect(topic.reload.closed).to eq(true)
         expect(timer.reload.execute_at).to eq_time(1.hour.from_now)
diff --git a/spec/models/topic_spec.rb b/spec/models/topic_spec.rb
index 829510d93a5..dfcb7ecb57f 100644
--- a/spec/models/topic_spec.rb
+++ b/spec/models/topic_spec.rb
@@ -1876,7 +1876,7 @@ describe Topic do
 
         freeze_time 3.hours.from_now
 
-        TopicTimer.ensure_consistency!
+        Jobs::TopicTimerEnqueuer.new.execute
         expect(topic.reload.closed).to eq(true)
       end
     end
diff --git a/spec/models/topic_timer_spec.rb b/spec/models/topic_timer_spec.rb
index 46bd16bd8dc..a9ad8b39bf7 100644
--- a/spec/models/topic_timer_spec.rb
+++ b/spec/models/topic_timer_spec.rb
@@ -10,6 +10,20 @@ RSpec.describe TopicTimer, type: :model do
   before { freeze_time }
 
   context "validations" do
+    describe "pending_timers scope" do
+      it "does not return deleted timers" do
+        topic_timer.trash!
+        expect(TopicTimer.pending_timers.pluck(:id)).not_to include(topic_timer.id)
+      end
+
+      it "does not return timers in the future of the provided before time" do
+        topic_timer.update!(execute_at: 3.days.from_now)
+        expect(TopicTimer.pending_timers.pluck(:id)).not_to include(topic_timer.id)
+        expect(TopicTimer.pending_timers(2.days.from_now).pluck(:id)).not_to include(topic_timer.id)
+        topic_timer.update!(execute_at: 1.minute.ago, created_at: 10.minutes.ago)
+        expect(TopicTimer.pending_timers.pluck(:id)).to include(topic_timer.id)
+      end
+    end
     describe '#status_type' do
       it 'should ensure that only one active public topic status update exists' do
         topic_timer.update!(topic: topic)
@@ -20,6 +34,44 @@ RSpec.describe TopicTimer, type: :model do
       end
     end
 
+    describe "#typed_job_scheduled?" do
+      let(:scheduled_jobs) { Sidekiq::ScheduledSet.new }
+
+      after do
+        scheduled_jobs.clear
+      end
+
+      it "returns true if the job is scheduled" do
+        Sidekiq::Testing.disable! do
+          scheduled_jobs.clear
+          Jobs.enqueue_at(3.hours.from_now, :close_topic, topic_timer_id: topic_timer.id)
+          expect(topic_timer.typed_job_scheduled?).to eq(true)
+        end
+      end
+
+      it "returns false if the job is not already scheduled" do
+        Sidekiq::Testing.disable! do
+          scheduled_jobs.clear
+          expect(topic_timer.typed_job_scheduled?).to eq(false)
+        end
+      end
+
+      it "returns true if the toggle_topic_closed job is scheduled for close,open,silent_close types" do
+        Sidekiq::Testing.disable! do
+          scheduled_jobs.clear
+          topic_timer1 = Fabricate(:topic_timer, status_type: TopicTimer.types[:close])
+          Jobs.enqueue_at(3.hours.from_now, :toggle_topic_closed, topic_timer_id: topic_timer1.id)
+          topic_timer2 = Fabricate(:topic_timer, status_type: TopicTimer.types[:open])
+          Jobs.enqueue_at(3.hours.from_now, :toggle_topic_closed, topic_timer_id: topic_timer2.id)
+          topic_timer3 = Fabricate(:topic_timer, status_type: TopicTimer.types[:silent_close])
+          Jobs.enqueue_at(3.hours.from_now, :toggle_topic_closed, topic_timer_id: topic_timer3.id)
+          expect(topic_timer1.typed_job_scheduled?).to eq(true)
+          expect(topic_timer2.typed_job_scheduled?).to eq(true)
+          expect(topic_timer3.typed_job_scheduled?).to eq(true)
+        end
+      end
+    end
+
     describe '#execute_at' do
       describe 'when #execute_at is greater than #created_at' do
         it 'should be valid' do
@@ -95,20 +147,7 @@ RSpec.describe TopicTimer, type: :model do
           :close_topic, topic_timer_id: topic_timer.id
         )
 
-        expect_enqueued_with(job: :close_topic, args: { topic_timer_id: topic_timer.id }, at: 3.days.from_now) do
-          topic_timer.update!(execute_at: 3.days.from_now, created_at: Time.zone.now)
-        end
-      end
-
-      describe 'when execute_at is smaller than the current time' do
-        it 'should enqueue the job immediately' do
-          expect_enqueued_with(job: :close_topic, args: { topic_timer_id: topic_timer.id }, at: Time.zone.now) do
-            topic_timer.update!(
-              execute_at: Time.zone.now - 1.hour,
-              created_at: Time.zone.now - 2.hour
-            )
-          end
-        end
+        topic_timer.update!(execute_at: 3.days.from_now, created_at: Time.zone.now)
       end
     end
 
@@ -121,9 +160,7 @@ RSpec.describe TopicTimer, type: :model do
           :close_topic, topic_timer_id: topic_timer.id
         )
 
-        expect_enqueued_with(job: :close_topic, args: { topic_timer_id: topic_timer.id }, at: topic_timer.execute_at) do
-          topic_timer.update!(user: admin)
-        end
+        topic_timer.update!(user: admin)
       end
     end
 
@@ -141,18 +178,9 @@ RSpec.describe TopicTimer, type: :model do
       end
 
       it 'should close the topic' do
-        topic_timer
+        topic_timer.send(:schedule_auto_open_job)
         expect(topic.reload.closed).to eq(true)
       end
-
-      describe 'when topic has been deleted' do
-        it 'should not queue the job' do
-          topic.trash!
-          topic_timer
-
-          expect(Jobs::ToggleTopicClosed.jobs).to eq([])
-        end
-      end
     end
 
     describe 'when a close topic status update is created for a closed topic' do
@@ -169,18 +197,9 @@ RSpec.describe TopicTimer, type: :model do
       end
 
       it 'should open the topic' do
-        topic_timer
+        topic_timer.send(:schedule_auto_close_job)
         expect(topic.reload.closed).to eq(false)
       end
-
-      describe 'when topic has been deleted' do
-        it 'should not queue the job' do
-          topic.trash!
-          topic_timer
-
-          expect(Jobs::ToggleTopicClosed.jobs).to eq([])
-        end
-      end
     end
 
     describe '#public_type' do
@@ -205,62 +224,6 @@ RSpec.describe TopicTimer, type: :model do
     end
   end
 
-  describe '.ensure_consistency!' do
-    it 'should enqueue jobs that have been missed' do
-      close_topic_timer = Fabricate(:topic_timer,
-        execute_at: Time.zone.now - 1.hour,
-        created_at: Time.zone.now - 2.hour
-      )
-
-      open_topic_timer = Fabricate(:topic_timer,
-        status_type: described_class.types[:open],
-        execute_at: Time.zone.now - 1.hour,
-        created_at: Time.zone.now - 2.hour,
-        topic: Fabricate(:topic, closed: true)
-      )
-
-      Fabricate(:topic_timer, execute_at: Time.zone.now + 1.hour)
-
-      trashed_close_topic_timer = Fabricate(:topic_timer,
-        execute_at: Time.zone.now - 1.hour,
-        created_at: Time.zone.now - 2.hour
-      )
-
-      trashed_close_topic_timer.topic.trash!
-
-      trashed_open_topic_timer = Fabricate(:topic_timer,
-        execute_at: Time.zone.now - 1.hour,
-        created_at: Time.zone.now - 2.hour,
-        status_type: described_class.types[:open]
-      )
-
-      trashed_open_topic_timer.topic.trash!
-
-      # creating topic timers already enqueues jobs
-      # let's delete them to test ensure_consistency!
-      Sidekiq::Worker.clear_all
-
-      expect { described_class.ensure_consistency! }
-        .to change { Jobs::CloseTopic.jobs.count }.by(2).and change { Jobs::OpenTopic.jobs.count }.by(2)
-
-      expect(job_enqueued?(job: :close_topic, args: {
-        topic_timer_id: close_topic_timer.id
-      })).to eq(true)
-
-      expect(job_enqueued?(job: :open_topic, args: {
-        topic_timer_id: open_topic_timer.id
-      })).to eq(true)
-
-      expect(job_enqueued?(job: :close_topic, args: {
-        topic_timer_id: trashed_close_topic_timer.id
-      })).to eq(true)
-
-      expect(job_enqueued?(job: :open_topic, args: {
-        topic_timer_id: trashed_open_topic_timer.id
-      })).to eq(true)
-    end
-  end
-
   describe "runnable?" do
     it "returns false if execute_at > now" do
       topic_timer = Fabricate.build(:topic_timer,
diff --git a/spec/support/sidekiq_helpers.rb b/spec/support/sidekiq_helpers.rb
index 815badebe73..12dce59e856 100644
--- a/spec/support/sidekiq_helpers.rb
+++ b/spec/support/sidekiq_helpers.rb
@@ -47,7 +47,7 @@ module SidekiqHelpers
   # end
   def expect_not_enqueued_with(job:, args: {}, at: nil)
     expect_enqueued_with(job: job, args: args, at: at, expectation: false) do
-      yield
+      yield if block_given?
     end
   end
 
@@ -61,6 +61,20 @@ module SidekiqHelpers
     match_jobs(jobs: klass.jobs, args: args, at: at)
   end
 
+  # Same as job_enqueued? except it checks the expectation is true.
+  # Use this if you need to check if more than one job is enqueued from
+  # a single command, unlike expect_enqueued_with which needs a block
+  # to run code for the expectation to work. E.g.
+  #
+  # expect_not_enqueued_with(job: :close_topic, args: { topic_timer_id: deleted_timer.id })
+  # expect_not_enqueued_with(job: :close_topic, args: { topic_timer_id: future_timer.id })
+  # subject.execute
+  # expect_job_enqueued(job: :close_topic, args: { topic_timer_id: timer1.id })
+  # expect_job_enqueued(job: :open_topic, args: { topic_timer_id: timer2.id })
+  def expect_job_enqueued(job:, args: {}, at: nil)
+    expect(job_enqueued?(job: job, args: args, at: at)).to eq(true)
+  end
+
   private
 
   def match_jobs(jobs:, args:, at:)