Prevent Duplicates with Delayed Jobs

add_signature_fields_to_delayed_jobs.rb

class AddFieldsToDelayedJobs < ActiveRecord::Migration
  def change
    add_column :delayed_jobs, :signature, :string
    add_index :delayed_jobs, :signature
  end
end

delayed_duplicate_prevention_plugin.rb

class ActiveJob::QueueAdapters::DelayedJobAdapter::JobWrapper
  def object
    job_data['job_class'].constantize
  end

  def args
    job_data['arguments']
  end

  def method_name
    'perform'
  end
end

class DelayedDuplicatePreventionPlugin < Delayed::Plugin
  module SignatureConcern
    extend ActiveSupport::Concern

    included do
      before_validation :generate_signature, on: :create
      validate :must_be_unique
    end

    private

    def generate_signature
      job_class_name = payload_object.object.name
      method_name = payload_object.method_name
      arguments = payload_object.args

      # MyAwesomeJob.perform(123, "articles")
      self.signature = "#{job_class_name}.#{method_name}(#{arguments.to_s[1..-2]})"
    end

    def must_be_unique
      # we want to allow duplicates for failed and currently running jobs
      # for example: "after update, sync the record to external service (elastic search, CMS, API)
      return unless Delayed::Job.exists?(signature: signature, failed_at: nil, locked_at: nil, locked_by: nil)

      logger.warn "Found duplicate job #{signature}, ignoring..."
      errors.add(:base, 'This is a duplicate')
    end
  end
end

Delayed::Backend::ActiveRecord::Job.include DelayedDuplicatePreventionPlugin::SignatureConcern
Delayed::Worker.plugins << DelayedDuplicatePreventionPlugin

delayed_job.rb

# config/initializers/delayed_job.rb
require 'delayed_duplicate_prevention_plugin'

Delayed::Backend::ActiveRecord::Job.send(:include, DelayedDuplicatePreventionPlugin::SignatureConcern)
Delayed::Worker.plugins << DelayedDuplicatePreventionPlugin