Tidy up and DRY SolidQueue::Job.enqueue_all_active_jobs

Taking advantage of the new methods created in previous commits.

Author: rosa

app/models/solid_queue/job.rb

@@ -1,66 +1,74 @@
 # frozen_string_literal: true
 
-class SolidQueue::Job < SolidQueue::Record
-  include Executable
+module SolidQueue
+  class Job < Record
+    include Executable
 
-  if Gem::Version.new(Rails.version) >= Gem::Version.new("7.1")
-    serialize :arguments, coder: JSON
-  else
-    serialize :arguments, JSON
-  end
-
-  DEFAULT_PRIORITY = 0
-  DEFAULT_QUEUE_NAME = "default"
+    if Gem::Version.new(Rails.version) >= Gem::Version.new("7.1")
+      serialize :arguments, coder: JSON
+    else
+      serialize :arguments, JSON
+    end
 
-  class << self
-    def enqueue_all_active_jobs(active_jobs)
-      scheduled_jobs, immediate_jobs = active_jobs.partition(&:scheduled_at)
-      with_concurrency_limits, without_concurrency_limits = immediate_jobs.partition(&:concurrency_limited?)
+    class << self
+      def enqueue_all(active_jobs)
+        scheduled_jobs, immediate_jobs = active_jobs.partition(&:scheduled_at)
+        with_concurrency_limits, without_concurrency_limits = immediate_jobs.partition(&:concurrency_limited?)
 
-      with_concurrency_limits.each do |active_job|
-        enqueue_active_job(active_job)
+        schedule_all_at_once(scheduled_jobs)
+        enqueue_all_at_once(without_concurrency_limits)
+        enqueue_one_by_one(with_concurrency_limits)
       end
 
-      transaction do
-        job_rows = scheduled_jobs.map { |job| attributes_from_active_job(job) }
-        insert_all(job_rows)
-        inserted_jobs = where(active_job_id: scheduled_jobs.map(&:job_id))
-        SolidQueue::ScheduledExecution.create_all_from_jobs(inserted_jobs)
+      def schedule_all_at_once(active_jobs)
+        transaction do
+          inserted_jobs = create_all_from_active_jobs(active_jobs)
+          schedule_all(inserted_jobs)
+        end
       end
 
-      transaction do
-        job_rows = without_concurrency_limits.map { |job| attributes_from_active_job(job) }
-        insert_all(job_rows)
-        inserted_jobs = where(active_job_id: without_concurrency_limits.map(&:job_id))
-        SolidQueue::ReadyExecution.create_all_from_jobs(inserted_jobs)
+      def enqueue_all_at_once(active_jobs)
+        transaction do
+          inserted_jobs = create_all_from_active_jobs(active_jobs)
+          dispatch_all_at_once(inserted_jobs)
+        end
       end
-    end
 
-    def enqueue_active_job(active_job, scheduled_at: Time.current)
-      enqueue **attributes_from_active_job(active_job).reverse_merge(scheduled_at: scheduled_at)
-    end
+      def enqueue_one_by_one(active_jobs)
+        active_jobs.each { |active_job| enqueue(active_job) }
+      end
 
-    private
-      def enqueue(**kwargs)
-        create!(**kwargs).tap do
-          SolidQueue.logger.debug "[SolidQueue] Enqueued job #{kwargs}"
+      def enqueue(active_job, scheduled_at: Time.current)
+        create!(**attributes_from_active_job(active_job).reverse_merge(scheduled_at: scheduled_at)).tap do |job|
+          active_job.provider_job_id = job.id
         end
       end
 
-      def attributes_from_active_job(active_job)
-        {
-          queue_name: active_job.queue_name,
-          active_job_id: active_job.job_id,
-          priority: active_job.priority,
-          scheduled_at: active_job.scheduled_at,
-          class_name: active_job.class.name,
-          arguments: active_job.serialize,
-          concurrency_key: active_job.concurrency_key
-        }.compact.with_defaults(defaults)
-      end
+      private
+        DEFAULT_PRIORITY = 0
+        DEFAULT_QUEUE_NAME = "default"
 
-      def defaults
-        { queue_name: DEFAULT_QUEUE_NAME, priority: DEFAULT_PRIORITY }
-      end
+        def create_all_from_active_jobs(active_jobs)
+          job_rows = active_jobs.map { |job| attributes_from_active_job(job) }
+          insert_all(job_rows)
+          where(active_job_id: active_jobs.map(&:job_id))
+        end
+
+        def attributes_from_active_job(active_job)
+          {
+            queue_name: active_job.queue_name,
+            active_job_id: active_job.job_id,
+            priority: active_job.priority,
+            scheduled_at: active_job.scheduled_at,
+            class_name: active_job.class.name,
+            arguments: active_job.serialize,
+            concurrency_key: active_job.concurrency_key
+          }.compact.with_defaults(defaults)
+        end
+
+        def defaults
+          { queue_name: DEFAULT_QUEUE_NAME, priority: DEFAULT_PRIORITY }
+        end
+    end
   end
 end

app/models/solid_queue/job/executable.rb

@@ -20,6 +20,16 @@ module Executable
         scope :failed, -> { includes(:failed_execution).where.not(failed_execution: { id: nil }) }
       end
 
+      class_methods do
+        def dispatch_all_at_once(jobs)
+          ReadyExecution.create_all_from_jobs(jobs)
+        end
+
+        def schedule_all(jobs)
+          ScheduledExecution.create_all_from_jobs(jobs)
+        end
+      end
+
       %w[ ready claimed failed scheduled ].each do |status|
         define_method("#{status}?") { public_send("#{status}_execution").present? }
       end

lib/active_job/queue_adapters/solid_queue_adapter.rb

@@ -9,19 +9,15 @@ module QueueAdapters
     #   Rails.application.config.active_job.queue_adapter = :solid_queue
     class SolidQueueAdapter
       def enqueue(active_job) # :nodoc:
-        SolidQueue::Job.enqueue_active_job(active_job).tap do |job|
-          active_job.provider_job_id = job.id
-        end
+        SolidQueue::Job.enqueue(active_job)
       end
 
       def enqueue_at(active_job, timestamp) # :nodoc:
-        SolidQueue::Job.enqueue_active_job(active_job, scheduled_at: Time.at(timestamp)).tap do |job|
-          active_job.provider_job_id = job.id
-        end
+        SolidQueue::Job.enqueue(active_job, scheduled_at: Time.at(timestamp))
       end
 
       def enqueue_all(active_jobs) # :nodoc:
-        SolidQueue::Job.enqueue_all_active_jobs(active_jobs)
+        SolidQueue::Job.enqueue_all(active_jobs)
       end
     end
   end

test/models/solid_queue/job_test.rb

@@ -31,7 +31,7 @@ class NonOverlappingGroupedJob2 < NonOverlappingJob
     active_job = AddToBufferJob.new(1).set(priority: 8, queue: "test")
 
     assert_ready do
-      SolidQueue::Job.enqueue_active_job(active_job)
+      SolidQueue::Job.enqueue(active_job)
     end
 
     solid_queue_job = SolidQueue::Job.last
@@ -51,7 +51,7 @@ class NonOverlappingGroupedJob2 < NonOverlappingJob
     active_job = AddToBufferJob.new(1).set(priority: 8, queue: "test")
 
     assert_scheduled do
-      SolidQueue::Job.enqueue_active_job(active_job, scheduled_at: 5.minutes.from_now)
+      SolidQueue::Job.enqueue(active_job, scheduled_at: 5.minutes.from_now)
     end
 
     solid_queue_job = SolidQueue::Job.last