add fa user enquerer

app/lib/fa_user_enqueuer.rb

@@ -0,0 +1,55 @@
+class FaUserEnqueuer
+  include HasColorLogger
+
+  def initialize(start_at:, low_water_mark:, high_water_mark:)
+    @low_water_mark = low_water_mark
+    @high_water_mark = high_water_mark
+    raise if @high_water_mark <= @low_water_mark
+    @user_iterator = Enumerator.new do |e|
+      Domain::Fa::User.
+        where(scanned_follows_at: nil, state: "ok").
+        where("id >= ?", start_at).
+        find_each do |user|
+        e << user
+      end
+    end
+  end
+
+  def run_once
+    already_enqueued = enqueued_count
+    if already_enqueued <= @low_water_mark
+      to_enqueue = @high_water_mark - already_enqueued
+      logger.info("enqueuing #{to_enqueue.to_s.bold} more users - #{already_enqueued.to_s.bold} already enqueued")
+      rows = to_enqueue.times.map do
+        @user_iterator.next
+      end
+      ReduxApplicationRecord.transaction do
+        rows.each do |user|
+          Domain::Fa::Job::UserFollowsJob.perform_later({ user: user })
+          logger.info "enqueue #{user.url_name.bold} (#{user.id.to_s.bold})"
+        end
+      end
+      throw StopIteration if rows.empty?
+    else
+      logger.info(
+        "#{already_enqueued.to_s.bold} already enqueued (max #{@high_water_mark.to_s.bold}) - " +
+          "waiting to fall below #{@low_water_mark.to_s.bold}"
+      )
+      :sleep
+    end
+  end
+
+  private
+
+  def enqueued_count
+    if Rails.env.test?
+      return SpecUtil.enqueued_jobs(Domain::Fa::Job::UserFollowsJob).count
+    end
+
+    GoodJob::Job.where(
+      queue_name: "fa_user_follows",
+      finished_at: nil,
+      performed_at: nil,
+    ).count
+  end
+end

rake/fa.rake

@@ -17,6 +17,24 @@ namespace :fa do
     end
   end
 
+  desc "enqueue waiting users"
+  task :enqueue_waiting_users => [:set_logger_stdout, :environment] do |t, args|
+    start_at = (ENV["start_at"] || 0).to_i
+    low_water_mark = 20
+    high_water_mark = 100
+    poll_duration = 10
+
+    enqueuer = FaUserEnqueuer.new(
+      start_at: start_at,
+      low_water_mark: low_water_mark,
+      high_water_mark: high_water_mark,
+    )
+
+    loop do
+      sleep poll_duration if enqueuer.run_once == :sleep
+    end
+  end
+
   desc "Import existing FA posts"
   task :import_existing, [:start_at] => [:environment] do |t, args|
     batch_size = args[:batch_size]&.to_i || ENV["batch_size"]&.to_i