Class: Coil::TransactionalMessagesPeriodicJob

Inherits:
ApplicationJob show all
Defined in:
app/jobs/coil/transactional_messages_periodic_job.rb

Constant Summary collapse

ATTEMPTS_THRESHOLD = 
3

Instance Method Summary collapse

Instance Method Details

#perform ⇒ Object 



12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
 
# File 'app/jobs/coil/transactional_messages_periodic_job.rb', line 12

def perform
  q = Sidekiq::Queue.new(Coil.sidekiq_queue)
  t = Time.current - q.latency - TransactionalMessagesJob::MAX_DURATION

  # Identify distinct message types, their associated job types, and
  # the distinct keys for which we have unprocessed messages.
  #
  # Exclude very recent messages, since a TransactionalMessagesJob could
  # still be processing those.
  #
  # Exclude keys where a processor has already initiated several attempts,
  # since that's a strong indicator that automatic retries are in play.
  #
  # Then, enqueue the appropriate jobs.
  message_parent_class.select(:type).distinct.pluck(:type).each do |type|
    message_class = message_class_for(type)
    next unless message_class.present?
    job_class = message_class.new.job_class

    message_class
      .unprocessed(processor_name: job_class.name)
      .where(created_at: nil...t)
      .group(:key)
      .having("MAX(processor_attempts) < ?", ATTEMPTS_THRESHOLD)
      .pluck(:key)
      .each { |k| job_class.perform_async(k) }
  end
end