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>
170 lines
6.0 KiB
Ruby
170 lines
6.0 KiB
Ruby
namespace :youtube do
|
|
desc "Test pipeline YouTube (setup → RTMP test → relay copy → activate). SESSION_ID= opzionale"
|
|
task e2e: :environment do
|
|
abort "YOUTUBE_CLIENT_ID mancante" if ENV["YOUTUBE_CLIENT_ID"].blank?
|
|
|
|
session = resolve_e2e_session!
|
|
puts "Sessione: #{session.id} (#{session.match.team.name} vs #{session.match.opponent_name})"
|
|
puts "Stato iniziale: #{session.status} broadcast=#{session.youtube_broadcast_id.present?} key=#{session.stream_key.present?}"
|
|
|
|
if Youtube::ApiThrottle.rate_limited?
|
|
puts "ATTENZIONE: rate limit attivo, attendo 120s..."
|
|
sleep 120
|
|
clear_youtube_throttle!
|
|
end
|
|
|
|
unless session.youtube_broadcast_id.present? && session.stream_key.present?
|
|
puts "→ Setup broadcast YouTube..."
|
|
setup_with_retries!(session)
|
|
session.reload
|
|
abort "Setup fallito: #{session.youtube_broadcast_id.inspect} / key=#{session.stream_key.present?}" unless session.stream_key.present?
|
|
end
|
|
|
|
puts "→ Watch URL: #{session.youtube_watch_url}"
|
|
|
|
session.begin_connect! if session.may_begin_connect?
|
|
session.update!(status: "connecting") unless session.status.in?(%w[connecting live reconnecting])
|
|
|
|
puts "→ Simulo telefono (RTMP test pattern 60s)..."
|
|
push_test_rtmp!(session, duration: 60)
|
|
|
|
wait_for("publisher video su MediaMTX", 45) { Mediamtx::PublisherOnline.video_publishing?(session.reload) }
|
|
|
|
Youtube::LivePipeline.schedule!(session, force: true)
|
|
wait_for("relay YouTube attivo", 90) { Streams::YoutubeRelay.running?(session.id) }
|
|
wait_for("ingest YouTube active", 120) { youtube_ingest_active?(session) }
|
|
|
|
Youtube::LivePipeline.schedule_activate!(session, force: true, wait: 5.seconds)
|
|
|
|
result = nil
|
|
wait_for("broadcast live", 180) do
|
|
result = Youtube::BroadcastService.new(session.match.team).activate_broadcast!(session.youtube_broadcast_id)
|
|
%i[live transitioned].include?(result)
|
|
end
|
|
|
|
puts "Risultato activate: #{result}"
|
|
puts "Ingest: #{ingest_status(session)}"
|
|
puts "Relay: #{relay_summary(session)}"
|
|
puts session.terminal? ? "Sessione terminata." : "OK — YouTube verificato."
|
|
end
|
|
|
|
desc "Termina sessioni YouTube di test più vecchie di 2h"
|
|
task cleanup_test: :environment do
|
|
old = StreamSession.where(platform: "youtube", status: %w[connecting live reconnecting paused idle])
|
|
.where("created_at < ?", 2.hours.ago)
|
|
old.find_each do |s|
|
|
Sessions::Stop.new(s).call
|
|
puts "ended #{s.id}"
|
|
rescue StandardError => e
|
|
puts "skip #{s.id}: #{e.message}"
|
|
end
|
|
end
|
|
|
|
def clear_youtube_throttle!
|
|
Redis.new(url: ENV.fetch("REDIS_URL", "redis://localhost:6379/0")).del(Youtube::ApiThrottle::KEY)
|
|
end
|
|
|
|
def setup_with_retries!(session, max_attempts: 10)
|
|
max_attempts.times do |i|
|
|
clear_youtube_throttle!
|
|
|
|
Youtube::SetupBroadcast.call(session)
|
|
session.reload
|
|
return if session.stream_key.present?
|
|
rescue Youtube::SetupBroadcast::Error => e
|
|
raise unless e.message.include?("temporaneamente") || e.message.include?("rate limit")
|
|
raise if i >= max_attempts - 1
|
|
|
|
wait = 60 + (i * 15)
|
|
puts " rate limit Google, attendo #{wait}s (tentativo #{i + 2}/#{max_attempts})..."
|
|
sleep wait
|
|
end
|
|
end
|
|
|
|
def resolve_e2e_session!
|
|
if ENV["SESSION_ID"].present?
|
|
return StreamSession.find(ENV["SESSION_ID"])
|
|
end
|
|
|
|
user = User.find_by(email: ENV.fetch("E2E_USER_EMAIL", "coach@matchlivetv.test"))
|
|
abort "Utente E2E non trovato" unless user
|
|
|
|
team = user.teams.first || Team.first
|
|
match = team.matches.order(scheduled_at: :desc).first
|
|
match ||= team.matches.create!(
|
|
opponent_name: "E2E Test #{Time.current.strftime("%H:%M")}",
|
|
scheduled_at: 5.minutes.from_now,
|
|
location: "Test automatico"
|
|
)
|
|
|
|
Sessions::Create.new(
|
|
user: user,
|
|
match: match,
|
|
params: { platform: "youtube", privacy_status: "unlisted", youtube_channel: "platform" }
|
|
).call
|
|
end
|
|
|
|
def wait_for(label, timeout_sec)
|
|
deadline = Time.now + timeout_sec
|
|
loop do
|
|
if yield
|
|
puts " ✓ #{label}"
|
|
return true
|
|
end
|
|
break if Time.now >= deadline
|
|
|
|
sleep 3
|
|
print "."
|
|
end
|
|
puts "\n ✗ timeout: #{label}"
|
|
false
|
|
end
|
|
|
|
def push_test_rtmp!(session, duration:)
|
|
intake = "rtmp://mediamtx:1935/#{session.mediamtx_path_name}"
|
|
cmd = [
|
|
"ffmpeg", "-hide_banner", "-loglevel", "error",
|
|
"-re", "-f", "lavfi", "-i", "testsrc=size=1280x720:rate=30",
|
|
"-f", "lavfi", "-i", "sine=frequency=440:sample_rate=48000",
|
|
"-t", duration.to_s,
|
|
"-c:v", "libx264", "-preset", "veryfast", "-pix_fmt", "yuv420p", "-b:v", "2500k",
|
|
"-c:a", "aac", "-b:a", "128k",
|
|
"-f", "flv", intake
|
|
]
|
|
pid = Process.spawn(*cmd)
|
|
unless Process.wait2(pid)[1].success?
|
|
warn "RTMP test inviato con warning (exit non zero)"
|
|
end
|
|
end
|
|
|
|
def ingest_status(session)
|
|
client = Youtube::BroadcastService.new(session.match.team).send(:authorized_client)
|
|
b = client.list_live_broadcasts("status,contentDetails", id: session.youtube_broadcast_id).items&.first
|
|
return "n/a" unless b
|
|
|
|
sid = b.content_details.bound_stream_id
|
|
stream = client.list_live_streams("status", id: sid).items&.first
|
|
"#{b.status.life_cycle_status} / ingest=#{stream&.status&.stream_status}"
|
|
rescue StandardError => e
|
|
e.message
|
|
end
|
|
|
|
def relay_summary(session)
|
|
pid = Redis.new(url: ENV.fetch("REDIS_URL", "redis://localhost:6379/0"))
|
|
.get(format(Streams::YoutubeRelay::REDIS_KEY, session.id))
|
|
pid.present? ? "pid=#{pid} (sidekiq copy)" : "non attivo"
|
|
end
|
|
|
|
def youtube_ingest_active?(session)
|
|
client = Youtube::BroadcastService.new(session.match.team).send(:authorized_client)
|
|
b = client.list_live_broadcasts("status,contentDetails", id: session.youtube_broadcast_id).items&.first
|
|
return false unless b
|
|
|
|
sid = b.content_details.bound_stream_id
|
|
stream = client.list_live_streams("status", id: sid).items&.first
|
|
stream&.status&.stream_status == "active"
|
|
rescue StandardError
|
|
false
|
|
end
|
|
end
|