feat: Consolidate enqueue and enqueue_at to use enqueue_all under the hood

README.md

@@ -284,11 +284,13 @@ The following events will be emitted automatically by workers as jobs are reserv
 - **delayed.job.run** - an event measuring the duration of a job's execution
 - **delayed.job.error** - an event indicating that a job has errored and may be retried (no duration attached)
 - **delayed.job.failure** - an event indicating that a job has permanently failed (no duration attached)
-- **delayed.job.enqueue** - an event measuring the time it takes to enqueue a job
+- **delayed.job.enqueue** - an event measuring the time it takes to enqueue one or more jobs (fires once per `Delayed::Job.enqueue` call and once per `perform_all_later` / `enqueue_all` batch)
 - **delayed.worker.reserve_jobs** - an event measuring the duration of the job "pickup query"
 
-The "run", "error", "failure" and "enqueue" events will include a `:job` argument in the event's payload,
-providing access to the job instance.
+The "run", "error", and "failure" events will include a `:job` argument in the event's payload,
+providing access to the job instance. The "enqueue" event will include a `:jobs` array (along with a
+`:count`) — note that elements may be either `Delayed::Job` instances (when enqueued directly via
+`Delayed::Job.enqueue`) or `ActiveJob` instances (when enqueued via `perform_later` / `perform_all_later`).
 
 ```ruby
 ActiveSupport::Notifications.subscribe('delayed.job.run') do |*args|

lib/delayed/active_job_adapter.rb

@@ -7,26 +7,63 @@ def enqueue_after_transaction_commit?
     end
 
     def enqueue(job)
-      _enqueue(job)
+      enqueue_all([job])
+      job
     end
 
     def enqueue_at(job, timestamp)
-      _enqueue(job, run_at: Time.at(timestamp)) # rubocop:disable Rails/TimeZone
+      job.scheduled_at = Time.at(timestamp) # rubocop:disable Rails/TimeZone
+      enqueue_all([job])
+      job
+    end
+
+    def enqueue_all(jobs)
+      return 0 if jobs.empty?
+
+      assert_safe_to_enqueue!(jobs)
+
+      Delayed.lifecycle.run_callbacks(:enqueue, jobs) do
+        now = Delayed::Job.db_time_now
+        rows = jobs.map { |job| build_insert_row(job, now) }
+        result = Delayed::Job.insert_all(rows) # rubocop:disable Rails/SkipsModelValidations
+        assign_provider_job_ids(jobs, result) if Delayed::Job.connection.supports_insert_returning?
+      end
+
+      mark_successfully_enqueued(jobs)
+      jobs.size
     end
 
     private
 
-    def _enqueue(job, opts = {})
-      if enqueue_after_transaction_commit_enabled?(job)
+    def assert_safe_to_enqueue!(jobs)
+      if jobs.any? { |job| enqueue_after_transaction_commit_enabled?(job) }
         raise UnsafeEnqueueError, "The ':delayed' ActiveJob adapter is not compatible with enqueue_after_transaction_commit"
       end
+      unless Delayed::Worker.delay_jobs == true
+        raise UnsafeEnqueueError, "The ':delayed' ActiveJob adapter is not compatible with delay_jobs false"
+      end
+    end
 
-      opts.merge!({ queue: job.queue_name, priority: job.priority }.compact)
-        .merge!(job.provider_attributes || {})
+    def assign_provider_job_ids(jobs, result)
+      ids = result.rows.map(&:first)
+      jobs.zip(ids) { |job, id| job.provider_job_id = id }
+    end
 
-      Delayed::Job.enqueue(JobWrapper.new(job), opts).tap do |dj|
-        job.provider_job_id = dj.id
-      end
+    def mark_successfully_enqueued(jobs)
+      jobs.each { |job| job.successfully_enqueued = true if job.respond_to?(:successfully_enqueued=) }
+    end
+
+    def build_insert_row(job, now)
+      opts = { queue: job.queue_name, priority: job.priority }.compact
+      opts.merge!(job.provider_attributes || {})
+      opts[:run_at] = coerce_scheduled_at(job.scheduled_at) if job.scheduled_at
+
+      prepared = Delayed::Backend::JobPreparer.new(JobWrapper.new(job), opts).prepare
+      Delayed::Job.new(prepared).attributes.compact.merge('created_at' => now, 'updated_at' => now)
+    end
+
+    def coerce_scheduled_at(value)
+      value.is_a?(Numeric) ? Time.at(value) : value # rubocop:disable Rails/TimeZone
     end
 
     def enqueue_after_transaction_commit_enabled?(job)

lib/delayed/backend/base.rb

@@ -14,7 +14,7 @@ def enqueue(*args)
 
         def enqueue_job(options)
           new(options).tap do |job|
-            Delayed.lifecycle.run_callbacks(:enqueue, job) do
+            Delayed.lifecycle.run_callbacks(:enqueue, [job]) do
               job.hook(:enqueue)
               Delayed::Worker.delay_job?(job) ? job.save : job.invoke_job
             end

lib/delayed/lifecycle.rb

@@ -4,7 +4,7 @@ class InvalidCallback < RuntimeError; end
   class Lifecycle
     EVENTS = {
       execute: [nil],
-      enqueue: [:job],
+      enqueue: [:jobs],
       perform: %i(worker job),
       error: %i(worker job),
       failure: %i(worker job),

lib/delayed/plugins/instrumentation.rb

@@ -2,9 +2,9 @@ module Delayed
   module Plugins
     class Instrumentation < Plugin
       callbacks do |lifecycle|
-        lifecycle.around(:enqueue) do |job, *args, &block|
-          ActiveSupport::Notifications.instrument('delayed.job.enqueue', active_support_notifications_tags(job)) do
-            block.call(job, *args)
+        lifecycle.around(:enqueue) do |jobs, &block|
+          ActiveSupport::Notifications.instrument('delayed.job.enqueue', bulk_enqueue_tags(jobs)) do
+            block.call(jobs)
           end
         end
 
@@ -34,6 +34,16 @@ def self.active_support_notifications_tags(job)
           job: job,
         }
       end
+
+      def self.bulk_enqueue_tags(jobs)
+        {
+          count: jobs.size,
+          table: Delayed::Job.table_name,
+          database: Delayed::Job.database_name,
+          database_adapter: Delayed::Job.database_adapter_name,
+          jobs: jobs,
+        }
+      end
     end
   end
 end