MatchLiveTv/backend/app/jobs/youtube_broadcast_setup_job.rb

class YoutubeBroadcastSetupJob < ApplicationJob
  queue_as :default

  RUN_LOCK_KEY = "youtube_setup:running:%s"
  MAX_RATE_LIMIT_ATTEMPTS = 15
  RATE_LIMIT_WAITS = [30, 45, 60, 90, 120, 120, 120, 120, 120, 120, 120, 120, 120, 120, 120].freeze

  def perform(session_id, youtube_channel = nil, attempt = 1)
    session = StreamSession.find_by(id: session_id)
    return unless session&.platform == "youtube"
    return if session.terminal?
    return if session.youtube_broadcast_id.present? && session.stream_key.present?
    return unless acquire_run_lock!(session_id, attempt)

    Youtube::SetupBroadcast.call(session, youtube_channel: youtube_channel.presence)
    release_run_lock!(session_id)
  rescue Youtube::SetupBroadcast::Error => e
    if rate_limit?(e) && attempt < MAX_RATE_LIMIT_ATTEMPTS
      Youtube::ApiThrottle.mark_rate_limited!
      wait = RATE_LIMIT_WAITS[attempt - 1] || 120
      extend_run_lock!(session_id, wait + 90)
      Rails.logger.warn(
        "[YoutubeBroadcastSetupJob] rate limit session=#{session_id} attempt=#{attempt} retry_in=#{wait}s"
      )
      self.class.set(wait: wait.seconds).perform_later(session_id, youtube_channel, attempt + 1)
      return
    end
    release_run_lock!(session_id)
    raise
  end

  private

  def rate_limit?(error)
    msg = error.message.to_s
    msg.include?("rate limit") ||
      msg.include?("userRequestsExceedRateLimit") ||
      msg.include?("temporaneamente occupato")
  end

  def acquire_run_lock!(session_id, attempt)
    key = format(RUN_LOCK_KEY, session_id)
    if attempt > 1
      return redis.exists?(key).positive?
    end

    redis.set(key, "1", nx: true, ex: 90)
  end

  def extend_run_lock!(session_id, seconds)
    redis.expire(format(RUN_LOCK_KEY, session_id), seconds)
  end

  def release_run_lock!(session_id)
    redis.del(format(RUN_LOCK_KEY, session_id))
  end

  def redis
    @redis ||= Redis.new(url: ENV.fetch("REDIS_URL", "redis://localhost:6379/0"))
  end
end