134 lines
3.7 KiB
Ruby
134 lines
3.7 KiB
Ruby
class Mailings::OutboundQueue
|
|
LOCK_PATH = Rails.root.join("tmp/outbound_smtp.lock")
|
|
STAMP_PATH = Rails.root.join("tmp/outbound_last_sent")
|
|
ATTEMPT_PATH = Rails.root.join("tmp/outbound_last_attempt")
|
|
|
|
class << self
|
|
attr_accessor :min_interval, :fail_interval
|
|
|
|
def enqueue(recipient)
|
|
recipient.with_lock do
|
|
return false unless recipient.mailing.sending?
|
|
return false unless recipient.pending?
|
|
|
|
recipient.update!(status: "queued", error_message: nil)
|
|
true
|
|
end
|
|
end
|
|
|
|
def drain_one!
|
|
File.open(lock_path, File::RDWR | File::CREAT, 0o644) do |file|
|
|
return :busy unless file.flock(File::LOCK_EX | File::LOCK_NB)
|
|
|
|
wait = seconds_until_next_slot
|
|
return :wait if wait.positive?
|
|
|
|
recipient = next_queued
|
|
return :empty if recipient.nil?
|
|
return :closed unless Mailings::SendClock.new(recipient.mailing).open?
|
|
|
|
begin
|
|
CampaignMailer.raise_delivery_errors = true
|
|
outcome = recipient.deliver_queued!
|
|
stamp_attempt!
|
|
stamp! if outcome == :sent
|
|
Rails.logger.info("[outbound] #{outcome} mailing=#{recipient.mailing_id} to=#{recipient.email}")
|
|
outcome == :sent ? :sent : :deferred
|
|
ensure
|
|
CampaignMailer.raise_delivery_errors = false if Rails.env.development?
|
|
end
|
|
end
|
|
end
|
|
|
|
def exclusive
|
|
File.open(lock_path, File::RDWR | File::CREAT, 0o644) do |file|
|
|
file.flock(File::LOCK_EX)
|
|
yield
|
|
end
|
|
end
|
|
|
|
def next_queued
|
|
# Sceglie il destinatario piu vecchio (per priorita/updated_at) TRA le mailing
|
|
# la cui finestra di invio e attualmente aperta. Prima si sceglieva il piu vecchio
|
|
# in assoluto: se apparteneva a una mailing con finestra chiusa, drain_one!
|
|
# restituiva :closed e si fermava, bloccando l'intera coda condivisa anche per
|
|
# le mailing senza vincoli orari (bug: coda bloccata da mailing con orario chiuso).
|
|
queued_scope.order(
|
|
Arel.sql(
|
|
"CASE WHEN mailing_recipients.error_message IS NULL OR mailing_recipients.error_message = '' THEN 0 ELSE 1 END, mailing_recipients.updated_at ASC, mailing_recipients.id ASC"
|
|
)
|
|
).includes(:mailing).find { |r| Mailings::SendClock.new(r.mailing).open? }
|
|
end
|
|
|
|
def queued_scope
|
|
MailingRecipient.queued.joins(:mailing).merge(Mailing.sending)
|
|
end
|
|
|
|
def seconds_until_next_slot
|
|
[
|
|
remaining(last_sent_at, min_interval),
|
|
remaining(last_attempt_at, fail_interval)
|
|
].max
|
|
end
|
|
|
|
def reset!
|
|
@last_sent_at = nil
|
|
@last_attempt_at = nil
|
|
File.delete(stamp_path) if File.exist?(stamp_path)
|
|
File.delete(attempt_path) if File.exist?(attempt_path)
|
|
rescue Errno::ENOENT
|
|
nil
|
|
end
|
|
|
|
def stamp!
|
|
@last_sent_at = Time.current
|
|
File.write(stamp_path, @last_sent_at.to_f.to_s)
|
|
end
|
|
|
|
def stamp_attempt!
|
|
@last_attempt_at = Time.current
|
|
File.write(attempt_path, @last_attempt_at.to_f.to_s)
|
|
end
|
|
|
|
def last_sent_at
|
|
@last_sent_at ||= read_time(stamp_path)
|
|
end
|
|
|
|
def last_attempt_at
|
|
@last_attempt_at ||= read_time(attempt_path)
|
|
end
|
|
|
|
private
|
|
|
|
def remaining(timestamp, interval)
|
|
return 0 if interval.to_f <= 0 || timestamp.nil?
|
|
|
|
leftover = interval - (Time.current - timestamp)
|
|
leftover.positive? ? leftover : 0
|
|
end
|
|
|
|
def read_time(path)
|
|
return unless File.exist?(path)
|
|
|
|
Time.zone.at(Float(File.read(path)))
|
|
rescue ArgumentError, TypeError, Errno::ENOENT
|
|
nil
|
|
end
|
|
|
|
def lock_path
|
|
LOCK_PATH
|
|
end
|
|
|
|
def stamp_path
|
|
STAMP_PATH
|
|
end
|
|
|
|
def attempt_path
|
|
ATTEMPT_PATH
|
|
end
|
|
end
|
|
|
|
self.min_interval = Rails.env.test? ? 0.0 : 60.0
|
|
self.fail_interval = Rails.env.test? ? 0.0 : 30.0
|
|
end
|