Pipeline YouTube con relay, overlay e publisher sync lato backend; fix GL pauseGlDuringRotation su Android per evitare crash in rotazione durante lo streaming Flutter. Co-authored-by: Cursor <cursoragent@cursor.com>
62 lines
1.9 KiB
Ruby
62 lines
1.9 KiB
Ruby
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
|