diff --git a/backend/app/controllers/api/v1/stream_sessions_controller.rb b/backend/app/controllers/api/v1/stream_sessions_controller.rb index c6b5449..d6c8fb4 100644 --- a/backend/app/controllers/api/v1/stream_sessions_controller.rb +++ b/backend/app/controllers/api/v1/stream_sessions_controller.rb @@ -56,6 +56,7 @@ module Api fps: params[:fps], last_seen_at: Time.current ) + sync_publisher_when_streaming!(params[:fps].to_f) SessionChannel.broadcast_message(@session, state.as_cable_payload) head :no_content end @@ -172,6 +173,19 @@ module Api occurred_at: event.occurred_at } end + + # L'app invia fps ogni ~10s; senza poll status.json l'overlay YouTube non partiva. + def sync_publisher_when_streaming!(fps) + return if fps < 1 + return if @session.terminal? || @session.paused? + return unless @session.platform == "youtube" + return unless @session.status.in?(%w[connecting live reconnecting]) + + Mediamtx::PublisherSync.new(@session).call + if @session.platform == "youtube" && @session.status.in?(%w[live connecting reconnecting]) + Mediamtx::PublisherSync.schedule_youtube_pipeline!(@session, force: false) + end + end end end end diff --git a/backend/app/controllers/concerns/mediamtx_playback.rb b/backend/app/controllers/concerns/mediamtx_playback.rb index aeb10c3..330de04 100644 --- a/backend/app/controllers/concerns/mediamtx_playback.rb +++ b/backend/app/controllers/concerns/mediamtx_playback.rb @@ -15,7 +15,7 @@ module MediamtxPlayback session.effective_hls_path_name end - # Player HLS: path _air con overlay quando il relay è attivo. + # Player HLS: path grezzo MediaMTX (live/match_{uuid}). def mediamtx_stream_playable?(session) return false if session.terminal? diff --git a/backend/app/controllers/public/live_controller.rb b/backend/app/controllers/public/live_controller.rb index b3ffdfc..ed8ca86 100644 --- a/backend/app/controllers/public/live_controller.rb +++ b/backend/app/controllers/public/live_controller.rb @@ -46,7 +46,10 @@ module Public @session = Mediamtx::PublisherSync.new(@session).call unless @session.terminal? @match = @session.match @stream_closed = @session.terminal? - @on_air = !@stream_closed && mediamtx_stream_playable?(@session) + @on_air = !@stream_closed && ( + mediamtx_stream_playable?(@session) || + @session.status.in?(%w[connecting live reconnecting paused]) + ) @publisher_online = !@stream_closed && mediamtx_publisher_online?(@session) end @@ -68,7 +71,7 @@ module Public session.live? || session.paused? || session.connecting? || session.reconnecting? || mediamtx_publisher_online?(session) ), - on_air: playable, + on_air: playable || (!closed && session.status.in?(%w[connecting live reconnecting paused])), publisher_online: publisher_online, awaiting_signal: !closed && !mediamtx_path_has_output?(session), showing_cover: !closed && mediamtx_path_has_output?(session) && diff --git a/backend/app/jobs/overlay_refresh_job.rb b/backend/app/jobs/overlay_refresh_job.rb index 0ed2ba6..3ef8b9d 100644 --- a/backend/app/jobs/overlay_refresh_job.rb +++ b/backend/app/jobs/overlay_refresh_job.rb @@ -27,9 +27,6 @@ class OverlayRefreshJob < ApplicationJob Streams::Overlay::Refresh.call(session) Mediamtx::PublisherSync.new(session).call if session.status.in?(%w[connecting reconnecting live paused]) - if session.platform == "youtube" - Streams::YoutubeRelay.ensure_publishing!(session) - end self.class.set(wait: INTERVAL).perform_later(session_id) end end diff --git a/backend/app/jobs/overlay_relay_ensure_job.rb b/backend/app/jobs/overlay_relay_ensure_job.rb new file mode 100644 index 0000000..6cbb07d --- /dev/null +++ b/backend/app/jobs/overlay_relay_ensure_job.rb @@ -0,0 +1,80 @@ +# Avvia overlay ffmpeg solo in Sidekiq (un solo job per sessione alla volta). +class OverlayRelayEnsureJob < ApplicationJob + queue_as :default + + LOCK_KEY = "overlay_ensure:lock:%s" + LOCK_TTL = 55 + RETRY_WAIT = 5.seconds + + class << self + def enqueue_unique(session_id, attempt: 1, force_restart: false, wait: 0.seconds) + key = format(LOCK_KEY, session_id) + r = Redis.new(url: ENV.fetch("REDIS_URL", "redis://localhost:6379/0")) + return unless r.set(key, "1", nx: true, ex: LOCK_TTL) + + set(wait: wait).perform_later(session_id, attempt, force_restart: force_restart) + end + end + + def perform(session_id, attempt = 1, force_restart: false) + session = StreamSession.find_by(id: session_id) + unless session && !session.terminal? + clear_lock(session_id) + return + end + return unless session.status.in?(%w[live connecting reconnecting paused]) + + begin + run_ensure(session, session_id, attempt, force_restart) + ensure + clear_lock(session_id) + end + end + + private + + def run_ensure(session, session_id, attempt, force_restart) + if Mediamtx::PublisherOnline.video_publishing?(session) && !session.paused? + session.go_live! if session.may_go_live? + session.reconnect! if session.reconnecting? && session.may_reconnect? + Mediamtx::Client.new.set_always_available(session, enabled: false) + end + + if force_restart + Streams::OverlayRelay.stop(session) + Streams::OverlayRelay.terminate_all_session_processes!(session.id) + end + + Streams::OverlayRelay.ensure_publishing!(session) + + if Streams::OverlayRelay.healthy?(session) + Youtube::LivePipeline.schedule_activate!(session, force: true, wait: 5.seconds) + return + end + + return if Streams::OverlayRelay.running?(session.id) + + Rails.logger.info("[OverlayRelayEnsureJob] overlay not running, retry session=#{session_id} attempt=#{attempt}") + reschedule(session_id, attempt, false) if attempt < 30 + rescue Mediamtx::Client::Error => e + Rails.logger.warn("[OverlayRelayEnsureJob] session=#{session_id}: #{e.message}") + end + + def reschedule(session_id, attempt, force_restart) + clear_lock(session_id) + self.class.enqueue_unique( + session_id, + attempt: attempt + 1, + force_restart: force_restart, + wait: RETRY_WAIT + ) + end + + def clear_lock(session_id) + redis.del(format(LOCK_KEY, session_id)) + end + + def redis + @redis ||= Redis.new(url: ENV.fetch("REDIS_URL", "redis://localhost:6379/0")) + end +end diff --git a/backend/app/jobs/overlay_score_refresh_job.rb b/backend/app/jobs/overlay_score_refresh_job.rb new file mode 100644 index 0000000..1c0b602 --- /dev/null +++ b/backend/app/jobs/overlay_score_refresh_job.rb @@ -0,0 +1,12 @@ +# Aggiorna tabellone overlay in background (evita blocchi HTTP al +1 punto). +class OverlayScoreRefreshJob < ApplicationJob + queue_as :default + + def perform(session_id) + session = StreamSession.find_by(id: session_id) + return unless session + return if session.terminal? + + Streams::Overlay::Refresh.call(session) + end +end diff --git a/backend/app/jobs/stream_publisher_sync_job.rb b/backend/app/jobs/stream_publisher_sync_job.rb new file mode 100644 index 0000000..e44b360 --- /dev/null +++ b/backend/app/jobs/stream_publisher_sync_job.rb @@ -0,0 +1,37 @@ +# Poll MediaMTX + pipeline YouTube automatica (webhook MediaMTX non disponibile in prod). +class StreamPublisherSyncJob < ApplicationJob + queue_as :default + + INTERVAL = 5.seconds + ACTIVE_STATUSES = %w[connecting live reconnecting paused].freeze + REDIS_CHAIN_KEY = "stream_publisher_sync:chain" + RUN_LOCK_KEY = "stream_publisher_sync:run_lock" + + def self.ensure_chain + return if redis.get(REDIS_CHAIN_KEY) + + redis.set(REDIS_CHAIN_KEY, "1", ex: INTERVAL.to_i * 4) + set(wait: INTERVAL).perform_later + end + + def self.redis + @redis ||= Redis.new(url: ENV.fetch("REDIS_URL", "redis://localhost:6379/0")) + end + + def perform + return unless self.class.redis.set(RUN_LOCK_KEY, "1", nx: true, ex: INTERVAL.to_i) + + begin + self.class.redis.set(REDIS_CHAIN_KEY, "1", ex: INTERVAL.to_i * 4) + + StreamSession + .where(status: ACTIVE_STATUSES) + .find_each do |session| + Mediamtx::PublisherSync.new(session).call + end + ensure + self.class.set(wait: INTERVAL).perform_later + self.class.redis.del(RUN_LOCK_KEY) + end + end +end diff --git a/backend/app/jobs/youtube_broadcast_activate_job.rb b/backend/app/jobs/youtube_broadcast_activate_job.rb index 2c1cca3..be43d40 100644 --- a/backend/app/jobs/youtube_broadcast_activate_job.rb +++ b/backend/app/jobs/youtube_broadcast_activate_job.rb @@ -1,32 +1,54 @@ class YoutubeBroadcastActivateJob < ApplicationJob queue_as :default - MAX_ATTEMPTS = 36 - RETRY_WAIT = 10.seconds - class << self - def schedule_with_retries(session_id, attempt: 1) - set(wait: attempt == 1 ? 8.seconds : RETRY_WAIT).perform_later(session_id, attempt) - end - end + MAX_ATTEMPTS = 30 + RETRY_WAIT = 15.seconds + RATE_LIMIT_WAIT = 90.seconds + LIVE_STATUSES = %w[live connecting reconnecting paused].freeze def perform(session_id, attempt = 1) session = StreamSession.find_by(id: session_id) return unless session&.platform == "youtube" return if session.terminal? return if session.youtube_broadcast_id.blank? - return unless Streams::YoutubeRelay.running?(session.id) + return unless LIVE_STATUSES.include?(session.status) + return unless Youtube::LivePipeline.broadcast_ready?(session) - result = Youtube::BroadcastService.new(session.match.team).activate_broadcast!(session.youtube_broadcast_id) - return if result == :live || result == :transitioned + unless Streams::YoutubeRelay.running?(session.id) + Streams::YoutubeRelay.ensure_publishing!(session) + return reschedule(session_id, attempt + 1) if attempt < MAX_ATTEMPTS - if result == :broadcast_ended - Youtube::SetupBroadcast.recreate!(session.reload) return end - return unless result == :waiting_ingest + result = Youtube::BroadcastService.new(session.match.team).activate_broadcast!(session.youtube_broadcast_id) + Rails.logger.info("[YoutubeBroadcastActivate] session=#{session.id} attempt=#{attempt} result=#{result}") + + return if result == :live || result == :transitioned + + if result == :rate_limited + reschedule(session_id, attempt, wait: RATE_LIMIT_WAIT) + return + end + + if result == :broadcast_ended + Youtube::SetupBroadcast.recreate!(session.reload) + Youtube::LivePipeline.schedule!(session.reload, force: true) + return + end + + if result == :waiting_ingest && (attempt % 4).zero? + Streams::YoutubeRelay.ensure_publishing!(session) + end + return if attempt >= MAX_ATTEMPTS - self.class.schedule_with_retries(session_id, attempt: attempt + 1) + reschedule(session_id, attempt + 1) + end + + private + + def reschedule(session_id, attempt, wait: RETRY_WAIT) + self.class.set(wait: wait).perform_later(session_id, attempt) end end diff --git a/backend/app/jobs/youtube_broadcast_setup_job.rb b/backend/app/jobs/youtube_broadcast_setup_job.rb index ffa752e..b3a2a50 100644 --- a/backend/app/jobs/youtube_broadcast_setup_job.rb +++ b/backend/app/jobs/youtube_broadcast_setup_job.rb @@ -1,12 +1,61 @@ class YoutubeBroadcastSetupJob < ApplicationJob queue_as :default - retry_on Youtube::SetupBroadcast::Error, wait: 10.seconds, attempts: 3 + 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) + def perform(session_id, youtube_channel = nil, attempt = 1) session = StreamSession.find_by(id: session_id) - return unless session + 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 diff --git a/backend/app/jobs/youtube_ingest_watchdog_job.rb b/backend/app/jobs/youtube_ingest_watchdog_job.rb new file mode 100644 index 0000000..af1e6a5 --- /dev/null +++ b/backend/app/jobs/youtube_ingest_watchdog_job.rb @@ -0,0 +1,57 @@ +# Controlli ripetuti finché YouTube non ha ingest active (poi attiva la broadcast). +class YoutubeIngestWatchdogJob < ApplicationJob + queue_as :default + + FIRST_WAIT = 20.seconds + NEXT_WAITS = [35.seconds, 45.seconds, 60.seconds].freeze + + def perform(session_id, phase = 0) + session = StreamSession.find_by(id: session_id) + return unless Youtube::LivePipeline.broadcast_ready?(session) + return if Youtube::ApiThrottle.rate_limited? + + ingest = youtube_ingest_status(session) + if ingest == "active" + Youtube::LivePipeline.schedule_activate!(session, force: true, wait: 3.seconds) + return + end + + if Streams::YoutubeRelay.running?(session.id) + Rails.logger.info("[YoutubeIngestWatchdog] relay ok ingest=#{ingest} phase=#{phase} session=#{session.id}") + else + Rails.logger.warn("[YoutubeIngestWatchdog] avvio relay ingest=#{ingest} session=#{session.id}") + Streams::YoutubeRelay.ensure_publishing!(session) + end + + schedule_next(session_id, phase) + rescue StandardError => e + Rails.logger.warn("[YoutubeIngestWatchdog] session=#{session_id}: #{e.message}") + schedule_next(session_id, phase) + end + + def self.schedule_chain(session_id) + set(wait: FIRST_WAIT).perform_later(session_id, 0) + end + + private + + def schedule_next(session_id, phase) + wait = NEXT_WAITS[phase] + return unless wait + + self.class.set(wait: wait).perform_later(session_id, phase + 1) + end + + def youtube_ingest_status(session) + return nil if session.youtube_broadcast_id.blank? + + 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 nil unless b + + sid = b.content_details.bound_stream_id + client.list_live_streams("status", id: sid).items.first&.status&.stream_status + rescue StandardError + nil + end +end diff --git a/backend/app/jobs/youtube_relay_ensure_job.rb b/backend/app/jobs/youtube_relay_ensure_job.rb new file mode 100644 index 0000000..3761657 --- /dev/null +++ b/backend/app/jobs/youtube_relay_ensure_job.rb @@ -0,0 +1,11 @@ +# Avvia/riavvia il relay YouTube solo nel container sidekiq (YOUTUBE_RELAY_WORKER=1). +class YoutubeRelayEnsureJob < ApplicationJob + queue_as :default + + def perform(session_id) + session = StreamSession.find_by(id: session_id) + return unless session + + Streams::YoutubeRelay.ensure_on_worker!(session) + end +end diff --git a/backend/app/jobs/youtube_relay_stop_job.rb b/backend/app/jobs/youtube_relay_stop_job.rb new file mode 100644 index 0000000..202559d --- /dev/null +++ b/backend/app/jobs/youtube_relay_stop_job.rb @@ -0,0 +1,10 @@ +class YoutubeRelayStopJob < ApplicationJob + queue_as :default + + def perform(session_id) + session = StreamSession.find_by(id: session_id) + return unless session + + Streams::YoutubeRelay.stop_on_worker!(session) if Streams::YoutubeRelay.worker? + end +end diff --git a/backend/app/models/stream_session.rb b/backend/app/models/stream_session.rb index 17afa11..425a1fd 100644 --- a/backend/app/models/stream_session.rb +++ b/backend/app/models/stream_session.rb @@ -18,7 +18,7 @@ class StreamSession < ApplicationRecord before_validation :normalize_privacy_status before_validation :ensure_publish_token, on: :create - scope :broadcasting, -> { where(status: %w[live connecting reconnecting]) } + scope :broadcasting, -> { where(status: %w[live connecting reconnecting paused]) } scope :publicly_listed, -> { where(privacy_status: "public") } scope :search_by_team_or_opponent, lambda { |query| q = query.to_s.strip @@ -95,18 +95,9 @@ class StreamSession < ApplicationRecord end def effective_hls_path_name - return mediamtx_path_name if terminal? - return mediamtx_overlay_path_name unless Streams::OverlayRelay.running?(id) - return mediamtx_overlay_path_name if overlay_air_has_output? - mediamtx_path_name end - def overlay_air_has_output? - info = Mediamtx::Client.new.list_paths.find { |i| i["name"] == mediamtx_overlay_path_name } - info && (info["online"] || info["ready"] || info["available"]) - end - def watch_page_url "#{MatchLiveTv.app_public_url.chomp('/')}/live/#{id}" end diff --git a/backend/app/services/mediamtx/client.rb b/backend/app/services/mediamtx/client.rb index 25489b3..1cd995c 100644 --- a/backend/app/services/mediamtx/client.rb +++ b/backend/app/services/mediamtx/client.rb @@ -31,28 +31,32 @@ module Mediamtx path = session.mediamtx_path_name # record: false finché non c'è publisher — con alwaysAvailable MediaMTX registrerebbe # solo la slate (schermo nero) in pausa/attesa. + # YouTube: niente slate sul path camera (maschera il video al relay ffmpeg). body = recording_body(session, enabled: false).merge( source: "publisher", - overridePublisher: true, - alwaysAvailable: true, - alwaysAvailableFile: slate_file_path + overridePublisher: true ) - # YouTube: relay continuo gestito da Streams::YoutubeRelay (ffmpeg in sidekiq/rails). + body[:alwaysAvailable] = true + body[:alwaysAvailableFile] = slate_file_path + # YouTube: telefono → MediaMTX; relay copy verso RTMPS in sidekiq. response = @conn.post("/v3/config/paths/add/#{CGI.escape(path)}", body) unless response.success? err = response.body.is_a?(Hash) ? response.body["error"] : response.body raise Error, "MediaMTX path create failed: #{response.status} #{err}" end + remember_always_available(path, enabled: true) true end def delete_path(session) delete_path_name(session.mediamtx_path_name) delete_path_name(session.mediamtx_overlay_path_name) + forget_always_available(session.mediamtx_path_name) end def delete_path_name(path_name) @conn.post("/v3/config/paths/delete/#{CGI.escape(path_name)}") + forget_always_available(path_name) end def set_path_recording(session, enabled:) @@ -67,16 +71,26 @@ module Mediamtx end def patch_path(session) + set_always_available(session, enabled: true) + end + + # Slate alwaysAvailable: copertina sullo stesso path quando il telefono è offline. + # Disattivare quando il publisher è in onda; riattivare in pausa/disconnessione. + def set_always_available(session, enabled:) path = session.mediamtx_path_name - body = { - alwaysAvailable: true, - alwaysAvailableFile: slate_file_path - } + return true if always_available_remembered?(path, enabled: enabled) + + body = if enabled + { alwaysAvailable: true, alwaysAvailableFile: slate_file_path } + else + { alwaysAvailable: false } + end response = @conn.patch("/v3/config/paths/patch/#{CGI.escape(path)}", body) unless response.success? err = response.body.is_a?(Hash) ? response.body["error"] : response.body - raise Error, "MediaMTX path patch failed: #{response.status} #{err}" + raise Error, "MediaMTX alwaysAvailable patch failed: #{response.status} #{err}" end + remember_always_available(path, enabled: enabled) true end @@ -152,5 +166,35 @@ module Mediamtx def slate_file_path ENV.fetch("MEDIAMTX_SLATE_FILE", "/slates/offline.mp4") end + + def always_available_remembered?(path, enabled:) + redis.get(always_available_key(path)) == always_available_value(enabled) + rescue Redis::BaseError + false + end + + def remember_always_available(path, enabled:) + redis.set(always_available_key(path), always_available_value(enabled), ex: 48.hours.to_i) + rescue Redis::BaseError + nil + end + + def forget_always_available(path) + redis.del(always_available_key(path)) + rescue Redis::BaseError + nil + end + + def always_available_key(path) + "mediamtx:always_available:#{path}" + end + + def always_available_value(enabled) + enabled ? "1" : "0" + end + + def redis + @redis ||= Redis.new(url: ENV.fetch("REDIS_URL", "redis://localhost:6379/0")) + end end end diff --git a/backend/app/services/mediamtx/publisher_online.rb b/backend/app/services/mediamtx/publisher_online.rb new file mode 100644 index 0000000..018d04a --- /dev/null +++ b/backend/app/services/mediamtx/publisher_online.rb @@ -0,0 +1,37 @@ +module Mediamtx + # Telefono connesso al path RTMP (non solo slate alwaysAvailable). + module PublisherOnline + module_function + + def active?(session) + active_path?(path_info(session)) + end + + def path_info(session) + Client.new.list_paths.find { |i| i["name"] == session.mediamtx_path_name } + end + + def active_path?(info) + return false unless info + + info["online"] == true && info.dig("source", "type") == "rtmpConn" + end + + def h264_video?(info) + return false unless info + + (info["tracks2"] || []).any? do |track| + track["codec"] == "H264" && track.dig("codecProps", "width").to_i.positive? + end + end + + def video_publishing?(session) + info = path_info(session) + return false unless active_path?(info) + # Slate alwaysAvailable ha H264 ma non è il telefono. + return false unless info.dig("source", "type") == "rtmpConn" + + h264_video?(info) + end + end +end diff --git a/backend/app/services/mediamtx/publisher_sync.rb b/backend/app/services/mediamtx/publisher_sync.rb index 931ab3a..4515818 100644 --- a/backend/app/services/mediamtx/publisher_sync.rb +++ b/backend/app/services/mediamtx/publisher_sync.rb @@ -10,40 +10,47 @@ module Mediamtx return @session if @session.terminal? info = Mediamtx::Client.new.list_paths.find { |i| i["name"] == @session.mediamtx_path_name } - publisher_online = publisher_online?(info) + publisher_online = Mediamtx::PublisherOnline.active_path?(info) if publisher_online + clear_publisher_misses!(@session.id) if @session.paused? # RTMP ancora connesso in pausa: non forzare live/reconnect. - elsif @session.may_go_live? - @session.go_live! - @session.reload - Streams::Overlay::Refresh.call(@session) - Streams::YoutubeRelay.ensure_publishing!(@session) if @session.platform == "youtube" - elsif @session.reconnecting? && @session.may_reconnect? - @session.reconnect! - cancel_timeout - Streams::Overlay::Refresh.call(@session.reload) + else + enable_live_path!(@session) + if @session.may_go_live? + @session.go_live! + @session.reload + kick_youtube!(@session) if @session.platform == "youtube" + elsif @session.reconnecting? && @session.may_reconnect? + @session.reconnect! + cancel_timeout + kick_youtube!(@session) + end + kick_youtube!(@session) if @session.platform == "youtube" end elsif !@session.paused? && @session.live? && @session.may_lose_connection? - @session.lose_connection! - schedule_timeout unless @session.paused? + if publisher_offline_sustained?(@session.id) + clear_publisher_misses!(@session.id) + restore_slate_path!(@session) + @session.lose_connection! + schedule_timeout unless @session.paused? + end + elsif !publisher_online && @session.status.in?(%w[reconnecting connecting]) && !@session.paused? + restore_slate_path!(@session) end - @session.reload + @session.reload.tap do |session| + schedule_youtube_relay_if_needed(session) + end end private - def publisher_online?(info) - return false unless info + def kick_youtube!(session) + return unless Youtube::LivePipeline.pipeline_eligible?(session) - return true if info["online"] - return true if info["source"].present? - # MediaMTX a volte lascia online=false con publisher attivo; i byte in ingresso confermano il segnale. - return true if info["ready"] && info["bytesReceived"].to_i > 1_000_000 - - false + Youtube::LivePipeline.tick_session!(session) end def cancel_timeout @@ -60,5 +67,55 @@ module Mediamtx ) @session.update!(timeout_job_id: job) end + + # Compatibilità con webhook / controller. + def self.schedule_youtube_pipeline!(session, force: false) + Youtube::LivePipeline.schedule!(session, force: force) + end + + # Match Live TV: lascia alwaysAvailable attivo — MediaMTX usa il publisher quando c'è + # e la copertina nei gap RTMP (telefono instabile). Disattivare la slate lasciava HLS + # senza stream ("no stream is available") e il player web in pausa/spinner. + def enable_live_path!(session) + return unless session.platform == "youtube" + + Client.new.set_always_available(session, enabled: false) + rescue Client::Error => e + Rails.logger.warn("[PublisherSync] disable slate session=#{session.id}: #{e.message}") + end + + def restore_slate_path!(session) + return if session.platform == "matchlivetv" + + Client.new.set_always_available(session, enabled: true) + rescue Client::Error => e + Rails.logger.warn("[PublisherSync] enable slate session=#{session.id}: #{e.message}") + end + + # Evita reconnect su un singolo poll offline (RTMP instabile dal telefono). + def publisher_offline_sustained?(session_id) + misses = redis.incr(format("publisher_offline_miss:%s", session_id)).to_i + redis.expire(format("publisher_offline_miss:%s", session_id), 120) + misses >= 4 + end + + def clear_publisher_misses!(session_id) + redis.del(format("publisher_offline_miss:%s", session_id)) + end + + def schedule_youtube_relay_if_needed(session) + return unless session.platform == "youtube" + return unless session.status.in?(%w[connecting live reconnecting paused]) + return unless Youtube::LivePipeline.broadcast_ready?(session) + + key = format("youtube_relay:sched:%s", session.id) + return unless redis.set(key, "1", nx: true, ex: 10) + + YoutubeRelayEnsureJob.perform_later(session.id) + end + + def redis + @redis ||= Redis.new(url: ENV.fetch("REDIS_URL", "redis://localhost:6379/0")) + end end end diff --git a/backend/app/services/scoring/apply_action.rb b/backend/app/services/scoring/apply_action.rb index 4a0a9e3..74a8d21 100644 --- a/backend/app/services/scoring/apply_action.rb +++ b/backend/app/services/scoring/apply_action.rb @@ -105,7 +105,6 @@ module Scoring def broadcast(score) SessionChannel.broadcast_message(@session, score.as_cable_payload) - Streams::Overlay::Refresh.call(@session) end end end diff --git a/backend/app/services/scoring/sync_state.rb b/backend/app/services/scoring/sync_state.rb index db30df9..7a90799 100644 --- a/backend/app/services/scoring/sync_state.rb +++ b/backend/app/services/scoring/sync_state.rb @@ -21,7 +21,6 @@ module Scoring score.update!(attrs) SessionChannel.broadcast_message(@session, score.as_cable_payload) - Streams::Overlay::Refresh.call(@session) score end end diff --git a/backend/app/services/sessions/create.rb b/backend/app/services/sessions/create.rb index 74a86bc..9d9e69b 100644 --- a/backend/app/services/sessions/create.rb +++ b/backend/app/services/sessions/create.rb @@ -38,8 +38,6 @@ module Sessions session.create_score_state! mtx = Mediamtx::Client.new mtx.create_path(session) - mtx.create_overlay_path(session) - start_overlay_relay(session) log_event(session, "pairing", { created: true, platform: session.platform }) end @@ -66,12 +64,6 @@ module Sessions Youtube::SetupBroadcast.call(session, youtube_channel: youtube_channel) end - def start_overlay_relay(session) - Streams::OverlayRelay.start(session) - rescue Streams::OverlayRelay::Error => e - Rails.logger.warn("[Sessions::Create] Overlay relay: #{e.message}") - end - def log_event(session, type, metadata) session.stream_events.create!(event_type: type, metadata: metadata, occurred_at: Time.current) end diff --git a/backend/app/services/sessions/pause.rb b/backend/app/services/sessions/pause.rb index e4e3a2b..534ad39 100644 --- a/backend/app/services/sessions/pause.rb +++ b/backend/app/services/sessions/pause.rb @@ -7,11 +7,12 @@ module Sessions def call cancel_timeout_job @session.pause! if @session.may_pause? - # Slate già su path (create_path + alwaysAvailable): stop RTMP basta, senza patch API. + Mediamtx::Client.new.set_always_available(@session, enabled: true) + # Slate su path per HLS in pausa; RTMP telefono si ferma via comando app. log_event("paused") SessionChannel.broadcast_message(@session, { type: "command", action: "pause_stream" }) SessionChannel.broadcast_message(@session, { type: "stream_event", event: "paused" }) - Streams::Overlay::Refresh.call(@session) + Youtube::LivePipeline.schedule!(@session, force: true) if @session.platform == "youtube" @session end diff --git a/backend/app/services/sessions/resume.rb b/backend/app/services/sessions/resume.rb index c201c5a..2da9fee 100644 --- a/backend/app/services/sessions/resume.rb +++ b/backend/app/services/sessions/resume.rb @@ -16,7 +16,7 @@ module Sessions log_event("resumed") SessionChannel.broadcast_message(@session, { type: "command", action: "resume_stream" }) SessionChannel.broadcast_message(@session, { type: "stream_event", event: "resumed" }) - Streams::Overlay::Refresh.call(@session) + Youtube::LivePipeline.schedule!(@session, force: true) if @session.platform == "youtube" @session end diff --git a/backend/app/services/sessions/start.rb b/backend/app/services/sessions/start.rb index 864f05f..583a5e5 100644 --- a/backend/app/services/sessions/start.rb +++ b/backend/app/services/sessions/start.rb @@ -8,6 +8,7 @@ module Sessions @session.match.team.entitlements.assert_concurrent_stream!(excluding_session: @session) @session.begin_connect! if @session.may_begin_connect? @session.update!(status: "connecting") unless @session.connecting? + Youtube::LivePipeline.schedule!(@session, force: true) if @session.platform == "youtube" broadcast_status("connecting") @session end diff --git a/backend/app/services/sessions/stop.rb b/backend/app/services/sessions/stop.rb index 662ef6b..e75357f 100644 --- a/backend/app/services/sessions/stop.rb +++ b/backend/app/services/sessions/stop.rb @@ -6,7 +6,6 @@ module Sessions def call cancel_timeout_job - Streams::OverlayRelay.stop(@session) Streams::YoutubeRelay.stop(@session) Sessions::RegiaAccess.revoke!(@session) @session.end_stream! diff --git a/backend/app/services/streams/overlay.rb b/backend/app/services/streams/overlay.rb index 29b6018..90ddceb 100644 --- a/backend/app/services/streams/overlay.rb +++ b/backend/app/services/streams/overlay.rb @@ -14,6 +14,17 @@ module Streams overlay_dir(session.id).join("overlay.pipe") end + # Logo chiaro watermark (brand/logo-white-m.png → branding/ in deploy). + def brand_logo_path + candidates = [ + Rails.root.join("branding/logo-white-m.png"), + Rails.root.join("brand/logo-white-m.png"), + Rails.root.join("public/logo-white.png"), + Rails.root.join("public/logo.png") + ] + candidates.find(&:file?) || candidates.first + end + def badge_away_color(team) club = team.club candidate = club&.effective_secondary_color.presence || "#1565c0" diff --git a/backend/app/services/streams/overlay/badge_label.rb b/backend/app/services/streams/overlay/badge_label.rb index 1fb15ec..20c5a55 100644 --- a/backend/app/services/streams/overlay/badge_label.rb +++ b/backend/app/services/streams/overlay/badge_label.rb @@ -16,7 +16,7 @@ module Streams return Result.new(text: "DIRETTA CHIUSA", background: "#374151", foreground: "#dddddd") if @session.terminal? return Result.new(text: "IN PAUSA", background: "#455a64", foreground: "#ffffff") if @session.paused? - if publisher_online? || (@session.live? && !@session.paused?) + if Mediamtx::PublisherOnline.active?(@session) || (@session.live? && !@session.paused?) return Result.new(text: "IN ONDA", background: "#2e7d32", foreground: "#ffffff") end @@ -27,7 +27,8 @@ module Streams if @session.connecting? return Result.new(text: "CONNECTING", background: "#f57c00", foreground: "#ffffff") end - return Result.new(text: "COPERTINA", background: "#455a64", foreground: "#ffffff") + # Slate alwaysAvailable senza telefono: non bruciare «COPERTINA» su YouTube. + return Result.new(text: "IN ATTESA", background: "#455a64", foreground: "#ffffff") end Result.new(text: "IN ATTESA", background: "#455a64", foreground: "#ffffff") @@ -36,12 +37,7 @@ module Streams private def path_info - @path_info ||= Mediamtx::Client.new.list_paths.find { |i| i["name"] == @session.mediamtx_path_name } - end - - def publisher_online? - info = path_info - info && info["online"] + @path_info ||= Mediamtx::PublisherOnline.path_info(@session) end def path_has_output? diff --git a/backend/app/services/streams/overlay/svg_builder.rb b/backend/app/services/streams/overlay/svg_builder.rb index 86b4329..0eae9b6 100644 --- a/backend/app/services/streams/overlay/svg_builder.rb +++ b/backend/app/services/streams/overlay/svg_builder.rb @@ -1,6 +1,6 @@ module Streams module Overlay - # PNG 1280×720 trasparente: tabellone alto-sinistra, badge alto-destra, brand sotto tabellone. + # PNG 1280×720 trasparente: tabellone alto-sinistra, badge alto-destra, logo chiaro (logo-white-m) basso-destra. class SvgBuilder CANVAS_W = 1280 CANVAS_H = 720 @@ -17,9 +17,10 @@ module Streams cols = score_columns <<~SVG - + #{scorebug_svg(cols)} #{badge_svg} + #{brand_watermark_svg} SVG end @@ -113,6 +114,22 @@ module Streams SVG end + def brand_watermark_svg + path = Overlay.brand_logo_path + return "" unless path.file? + + w = 108 + h = 108 + x = CANVAS_W - 14 - w + y = CANVAS_H - 14 - h + uri = "file://#{path.expand_path}" + <<~SVG + + + + SVG + end + def badge_svg text = @badge.text bg = @badge.background diff --git a/backend/app/services/streams/overlay_relay.rb b/backend/app/services/streams/overlay_relay.rb index 92f13c1..8a1381c 100644 --- a/backend/app/services/streams/overlay_relay.rb +++ b/backend/app/services/streams/overlay_relay.rb @@ -4,92 +4,115 @@ module Streams class Error < StandardError; end REDIS_KEY = "overlay_relay:pid:%s" + LIVE_INTAKE_KEY = "overlay_relay:live_intake:%s" + MODE_KEY = "overlay_relay:mode:%s" + START_LOCK_KEY = "overlay_relay:start_lock:%s" + MODE_LIVE = "live".freeze + MODE_COVER = "cover".freeze class << self - def start(session) + def start(session, mode: nil) return if session.terminal? + return unless acquire_start_lock!(session.id) - stop(session) if pid_for(session.id).present? + begin + terminate_all_session_processes!(session.id) - Streams::Overlay::Refresh.call(session) + Streams::Overlay::Refresh.call(session) - dir = Streams::Overlay.overlay_dir(session.id) - FileUtils.mkdir_p(dir) - png = Streams::Overlay.overlay_png_path(session) - pipe = Streams::Overlay.overlay_pipe_path(session) - ensure_pipe!(pipe) - intake = "#{mediamtx_rtmp_url}/#{session.mediamtx_path_name}" - output = "#{mediamtx_rtmp_url}/#{session.mediamtx_overlay_path_name}" - log_path = log_file(session) - FileUtils.mkdir_p(File.dirname(log_path)) + dir = Streams::Overlay.overlay_dir(session.id) + FileUtils.mkdir_p(dir) + png = Streams::Overlay.overlay_png_path(session) + pipe = Streams::Overlay.overlay_pipe_path(session) + ensure_pipe!(pipe) + log_path = log_file(session) + FileUtils.mkdir_p(File.dirname(log_path)) - fps = output_fps - overlay_fps = overlay_input_fps - filter = "[1:v]format=rgba[ol];[0:v]fps=#{fps},format=yuv420p[vid];[vid][ol]overlay=0:0:format=auto[vout]" - - ffmpeg_pid = Process.spawn( - "ffmpeg", "-nostdin", "-hide_banner", "-loglevel", "warning", - "-fflags", "+genpts+discardcorrupt", - "-analyzeduration", "10000000", "-probesize", "10000000", - "-rw_timeout", "15000000", - "-i", intake, - "-f", "image2pipe", "-framerate", overlay_fps.to_s, "-i", pipe.to_s, - "-filter_complex", filter, - "-map", "[vout]", "-map", "0:a?", - "-fps_mode", "cfr", "-r", fps.to_s, - "-c:v", "libx264", "-preset", x264_preset, "-bf", "0", - "-profile:v", "high", "-pix_fmt", "yuv420p", - "-b:v", bitrate_for(session), "-maxrate", maxrate_for(session), "-bufsize", bufsize_for(session), - "-g", fps.to_s, "-keyint_min", fps.to_s, - "-c:a", "aac", "-b:a", "128k", "-ar", "48000", "-ac", "1", - "-f", "flv", output, - %i[out err] => log_path, - pgroup: true - ) - spawn_png_feeder!(ffmpeg_pid, pipe, png, log_path) - Process.detach(ffmpeg_pid) - store_pid(session.id, ffmpeg_pid) - OverlayRefreshJob.schedule_chain(session.id) - Rails.logger.info("[OverlayRelay] started pid=#{ffmpeg_pid} session=#{session.id}") - ffmpeg_pid + mode = normalize_mode(mode || desired_mode(session)) + ffmpeg_pid = Process.spawn(*ffmpeg_command(session, pipe, mode), %i[out err] => log_path, pgroup: true) + spawn_png_feeder!(ffmpeg_pid, pipe, png, log_path) + Process.detach(ffmpeg_pid) + store_pid(session.id, ffmpeg_pid) + store_mode(session.id, mode) + mode == MODE_LIVE ? mark_live_intake!(session.id) : clear_live_intake!(session.id) + OverlayRefreshJob.schedule_chain(session.id) + if session.platform == "youtube" && session.stream_key.present? + Streams::YoutubeRelay.stop(session) + Youtube::LivePipeline.schedule_activate!(session, force: true, wait: 20.seconds) + end + Rails.logger.info("[OverlayRelay] started pid=#{ffmpeg_pid} mode=#{mode} session=#{session.id} youtube_tee=#{session.platform == "youtube"}") + ffmpeg_pid + ensure + release_start_lock!(session.id) + end rescue Errno::ENOENT => e raise Error, "ffmpeg non disponibile: #{e.message}" end def stop(session) - pid = pid_for(session.id) - return false if pid.blank? + pids = session_process_pids(session.id) + redis_pid = pid_for(session.id).to_i + pids << redis_pid if redis_pid > 1 + pids = pids.uniq.select { |p| p > 1 } + return false if pids.empty? - terminate_pid(pid) + pids.each { |p| terminate_pid(p) } clear_pid(session.id) + clear_mode(session.id) + clear_live_intake!(session.id) pipe = Streams::Overlay.overlay_pipe_path(session) FileUtils.rm_f(pipe) if pipe.exist? OverlayRefreshJob.cancel_chain(session.id) - Rails.logger.info("[OverlayRelay] stopped pid=#{pid} session=#{session.id}") + Rails.logger.info("[OverlayRelay] stopped pids=#{pids.join(",")} session=#{session.id}") true end def running?(session_id) - pid = pid_for(session_id) - return false if pid.blank? + return false unless overlay_relay_process_check_local? - return true unless overlay_relay_process_check_local? + alive = session_process_pids(session_id).select { |p| process_alive?(p) } + return true if alive.any? - process_alive?(pid) + redis_pid = pid_for(session_id).to_i + return false if redis_pid <= 1 + + process_alive?(redis_pid) + end + + # Overlay ffmpeg vivo e path _air riceve dati. + def healthy?(session) + running?(session.id) && overlay_path_publishing?(session) end # Riavvia il relay se il processo è morto o _air non pubblica (es. feeder bloccato). def ensure_publishing!(session) return if session.terminal? + restart_if_missing_youtube_tee!(session) + wanted_mode = desired_mode(session) + pids = session_process_pids(session.id) + if pids.any? && pids.none? { |p| process_alive?(p) } + terminate_all_session_processes!(session.id) + end + if healthy?(session) && current_mode(session.id) == wanted_mode + Youtube::LivePipeline.schedule_activate!(session) if session.platform == "youtube" + return + end - if running?(session.id) && overlay_path_publishing?(session) + if running?(session.id) && current_mode(session.id) != wanted_mode + return unless mode_switch_allowed?(session.id, wanted_mode) + + Rails.logger.info("[OverlayRelay] switch mode #{current_mode(session.id)}→#{wanted_mode} session=#{session.id}") + stop(session) + end + + if running?(session.id) && overlay_path_publishing?(session) && current_mode(session.id) == wanted_mode return end if running?(session.id) misses = redis.incr(overlay_air_miss_key(session.id)).to_i redis.expire(overlay_air_miss_key(session.id), 120) - return if misses < 3 + return if misses < 4 redis.del(overlay_air_miss_key(session.id)) Rails.logger.warn("[OverlayRelay] _air non attivo dopo #{misses} tentativi, restart session=#{session.id}") @@ -98,24 +121,156 @@ module Streams redis.del(overlay_air_miss_key(session.id)) end - start(session) + start(session, mode: wanted_mode) rescue Error => e Rails.logger.warn("[OverlayRelay] ensure_publishing session=#{session.id}: #{e.message}") end + def desired_mode(session) + return MODE_COVER if session.paused? + return MODE_LIVE if Mediamtx::PublisherOnline.video_publishing?(session) + # RTMP a tratti: resta su live finché c'era segnale telefono (evita flip copertina↔live). + return MODE_LIVE if session.reconnecting? && live_intake?(session.id) + return MODE_LIVE if session.status.in?(%w[live connecting reconnecting]) && recent_phone_bytes?(session) + + MODE_COVER + end + + def terminate_all_session_processes!(session_id) + session_process_pids(session_id).each do |pid| + terminate_pid(pid) + end + redis_pid = pid_for(session_id).to_i + terminate_pid(redis_pid) if redis_pid > 1 + clear_pid(session_id) + clear_mode(session_id) + end + + def session_process_pids(session_id) + tag = session_match_tag(session_id) + pids = [] + each_overlay_process do |pid, cmdline| + pids << pid if cmdline.include?(tag) + end + pids.uniq + end + + def session_match_tag(session_id) + "match_#{session_id}" + end + + def each_overlay_process + return unless overlay_relay_process_check_local? + + Dir.glob("/proc/[0-9]*").each do |proc_dir| + pid = proc_dir.split("/").last.to_i + next if pid <= 1 + + cmdline = File.read("#{proc_dir}/cmdline").tr("\0", " ") + next unless cmdline.include?("ffmpeg") || cmdline.include?("overlay_png_feeder") + next unless cmdline.include?("match_") + + yield pid, cmdline + rescue Errno::ENOENT, Errno::EPERM + nil + end + end + private def mediamtx_rtmp_url ENV.fetch("MEDIAMTX_INTERNAL_RTMP_URL", "rtmp://mediamtx:1935").chomp("/") end + def ffmpeg_command(session, pipe, mode) + mode == MODE_LIVE ? live_command(session, pipe) : cover_command(session, pipe) + end + + def live_command(session, pipe) + intake = "#{mediamtx_rtmp_url}/#{session.mediamtx_path_name}" + [ + "ffmpeg", "-nostdin", "-hide_banner", "-loglevel", "warning", + "-fflags", "+genpts+discardcorrupt", + "-analyzeduration", "20000000", "-probesize", "20000000", + "-rw_timeout", "15000000", + "-noautorotate", + "-i", intake, + "-thread_queue_size", "512", + "-f", "image2pipe", "-framerate", overlay_input_fps.to_s, "-i", pipe.to_s, + *common_encode_args(session, live_filter, "0:a?") + ] + end + + def cover_command(session, pipe) + cover = cover_image_path + raise Error, "copertina fallback non trovata: #{cover}" unless File.exist?(cover) + + [ + "ffmpeg", "-nostdin", "-hide_banner", "-loglevel", "warning", + "-re", "-loop", "1", "-framerate", output_fps.to_s, "-i", cover, + "-thread_queue_size", "512", + "-f", "image2pipe", "-framerate", overlay_input_fps.to_s, "-i", pipe.to_s, + "-f", "lavfi", "-i", "anullsrc=channel_layout=mono:sample_rate=48000", + *common_encode_args(session, cover_filter, "2:a") + ] + end + + def common_encode_args(session, filter, audio_map) + fps = output_fps + [ + "-filter_complex", filter, + "-map", "[vout]", "-map", audio_map, + "-fps_mode", "cfr", "-r", fps.to_s, + "-c:v", "libx264", "-preset", x264_preset, "-bf", "0", + "-profile:v", "high", "-pix_fmt", "yuv420p", + "-b:v", bitrate_for(session), "-maxrate", maxrate_for(session), "-bufsize", bufsize_for(session), + "-g", fps.to_s, "-keyint_min", fps.to_s, + "-c:a", "aac", "-b:a", "128k", "-ar", "48000", "-ac", "1", + *output_args(session) + ] + end + + # Crop center 16:9 a tutto schermo (niente bande nere su TV/web). + def live_filter + fps = output_fps + "[1:v]format=rgba[ol];" \ + "[0:v]scale=1280:720:force_original_aspect_ratio=increase,crop=1280:720,setsar=1,fps=#{fps},format=yuv420p[vid];" \ + "[vid][ol]overlay=0:0:format=auto[vout]" + end + + def cover_filter + fps = output_fps + "[1:v]format=rgba[ol];" \ + "[0:v]scale=1280:720:force_original_aspect_ratio=increase,crop=1280:720,setsar=1,fps=#{fps},format=yuv420p[vid];" \ + "[vid][ol]overlay=0:0:format=auto[vout]" + end + + def cover_image_path + ENV.fetch( + "OVERLAY_COVER_IMAGE", + Rails.root.join("branding", "CopertinaCanale_6.png").to_s + ) + end + + # Un solo encode → MediaMTX _air + YouTube RTMPS (evita seconda lettura RTMP senza SPS). + def output_args(session) + air = "#{mediamtx_rtmp_url}/#{session.mediamtx_overlay_path_name}" + if session.platform == "youtube" && session.stream_key.present? + yt = "rtmps://a.rtmps.youtube.com/live2/#{session.stream_key}" + tee = "[f=flv:onfail=ignore]#{air}|[f=flv:onfail=ignore:use_fifo=1]#{yt}" + ["-f", "tee", tee] + else + ["-f", "flv", air] + end + end + # Ricodifica con overlay: serve più budget della sorgente (generational loss). def output_video_kbps(session) override = ENV["OVERLAY_RELAY_VIDEO_KBPS"].presence&.to_i return override if override&.positive? - source_kbps = [(session.target_bitrate.to_i / 1000), 2500].max - [(source_kbps * 1.8).to_i, 4500].max + source_kbps = [(session.target_bitrate.to_i / 1000), 3000].max + [(source_kbps * 2.0).to_i, 5500].max end def bitrate_for(session) @@ -166,7 +321,52 @@ module Streams def overlay_path_publishing?(session) info = Mediamtx::Client.new.list_paths.find { |i| i["name"] == session.mediamtx_overlay_path_name } - info && (info["online"] || info["ready"] || info["available"]) + info && info["online"] && info["bytesReceived"].to_i > 100_000 + end + + def acquire_start_lock!(session_id) + redis.set(format(START_LOCK_KEY, session_id), "1", nx: true, ex: 30) + end + + def release_start_lock!(session_id) + redis.del(format(START_LOCK_KEY, session_id)) + end + + def publisher_has_video?(session) + info = Mediamtx::PublisherOnline.path_info(session) + return false unless Mediamtx::PublisherOnline.active_path?(info) + + (info["tracks2"] || []).any? do |track| + track["codec"] == "H264" && track.dig("codecProps", "width").to_i.positive? + end + end + + def live_intake?(session_id) + redis.get(format(LIVE_INTAKE_KEY, session_id)) == "1" + end + + def current_mode(session_id) + redis.get(format(MODE_KEY, session_id)) + end + + def normalize_mode(mode) + mode == MODE_LIVE ? MODE_LIVE : MODE_COVER + end + + def mark_live_intake!(session_id) + redis.set(format(LIVE_INTAKE_KEY, session_id), "1", ex: 48.hours.to_i) + end + + def clear_live_intake!(session_id) + redis.del(format(LIVE_INTAKE_KEY, session_id)) + end + + def store_mode(session_id, mode) + redis.set(format(MODE_KEY, session_id), normalize_mode(mode), ex: 48.hours.to_i) + end + + def clear_mode(session_id) + redis.del(format(MODE_KEY, session_id)) end def log_file(session) @@ -193,6 +393,46 @@ module Streams format("overlay_relay:air_miss:%s", session_id) end + def recent_phone_bytes?(session) + info = Mediamtx::PublisherOnline.path_info(session) + info && info["bytesReceived"].to_i > 1_000_000 + end + + def publishing_to_youtube?(session) + return false unless session.platform == "youtube" && session.stream_key.present? + + session_process_pids(session.id).any? do |pid| + cmdline = File.read("/proc/#{pid}/cmdline").tr("\0", " ") + cmdline.include?("rtmps://") || cmdline.include?("youtube.com/live2") + rescue Errno::ENOENT, Errno::EPERM + false + end + end + + def restart_if_missing_youtube_tee!(session) + return unless session.platform == "youtube" && session.stream_key.present? + return unless running?(session.id) + return if publishing_to_youtube?(session) + + Rails.logger.warn("[OverlayRelay] restart overlay: manca tee YouTube session=#{session.id}") + stop(session) + end + + def mode_switch_allowed?(session_id, wanted_mode) + key = format("overlay_relay:mode_want:%s", session_id) + prev = redis.get(key) + redis.set(key, wanted_mode, ex: 120) + return true if prev.blank? || prev == wanted_mode + + ticks = redis.incr(format("overlay_relay:mode_ticks:%s", session_id)).to_i + redis.expire(format("overlay_relay:mode_ticks:%s", session_id), 60) + if ticks >= 3 + redis.del(format("overlay_relay:mode_ticks:%s", session_id)) + return true + end + false + end + def process_alive?(pid) stat = File.read("/proc/#{pid.to_i}/stat") return false if stat.split[2] == "Z" diff --git a/backend/app/services/streams/youtube_relay.rb b/backend/app/services/streams/youtube_relay.rb index 4a80099..dc18b0a 100644 --- a/backend/app/services/streams/youtube_relay.rb +++ b/backend/app/services/streams/youtube_relay.rb @@ -1,62 +1,63 @@ module Streams - # Relay continuo verso YouTube: legge l'uscita MediaMTX del path (camera o slate alwaysAvailable) - # senza interrompere FLV quando il telefono si mette in pausa o perde rete. + # Relay verso YouTube: legge RTMP/HLS da MediaMTX e inoltra su RTMPS (-c copy). Nessun overlay. + # ffmpeg gira solo nel container sidekiq (YOUTUBE_RELAY_WORKER=1). class YoutubeRelay class Error < StandardError; end REDIS_KEY = "youtube_relay:pid:%s" + OWNER_KEY = "youtube_relay:owner:%s" class << self + def worker? + ENV["YOUTUBE_RELAY_WORKER"] == "1" + end + def start(session) - return unless session.platform == "youtube" - return if session.stream_key.blank? - return if session.terminal? - - return pid_for(session.id).to_i if running?(session.id) - - stop(session) if pid_for(session.id).present? - - log_path = log_file(session) - FileUtils.mkdir_p(File.dirname(log_path)) - intake = mediamtx_intake_url(session) - output = "rtmp://a.rtmp.youtube.com/live2/#{session.stream_key}" - - # YouTube richiede FLV H.264/AAC stabili; -c copy lascia spesso streamStatus=inactive. - pid = Process.spawn( - *youtube_ffmpeg_args(intake, output), - %i[out err] => log_path, - pgroup: true - ) - Process.detach(pid) - store_pid(session.id, pid) - Rails.logger.info("[YoutubeRelay] started pid=#{pid} session=#{session.id}") - schedule_youtube_activate(session) - pid - rescue Errno::ENOENT => e - raise Error, "ffmpeg non disponibile: #{e.message}" + return unless worker? + start_on_worker!(session) end def stop(session) - pid = pid_for(session.id) - return false if pid.blank? - - terminate_pid(pid) clear_pid(session.id) - Rails.logger.info("[YoutubeRelay] stopped pid=#{pid} session=#{session.id}") + redis.del(format(OWNER_KEY, session.id)) + if worker? + stop_on_worker!(session) + else + YoutubeRelayStopJob.perform_later(session.id) + end true end def running?(session_id) + owner = redis.get(format(OWNER_KEY, session_id)) pid = pid_for(session_id) - pid.present? && process_alive?(pid) + return false if pid.blank? + + return process_alive?(pid) if owner.blank? || owner == worker_id + + # Relay avviato in un altro container: consideralo attivo se il lock è recente. + redis.ttl(format(OWNER_KEY, session_id)) > 30 end - # Riavvia il relay se è morto (es. avviato prima del segnale su _air). def ensure_publishing!(session) return unless session.platform == "youtube" return if session.terminal? return if session.stream_key.blank? - return unless Streams::OverlayRelay.running?(session.id) + + if worker? + ensure_on_worker!(session) + else + YoutubeRelayEnsureJob.perform_later(session.id) + end + end + + def ensure_on_worker!(session) + return unless worker? + return unless session.platform == "youtube" + return if session.terminal? + return if session.stream_key.blank? + return unless session.status.in?(%w[live connecting reconnecting paused]) + return unless intake_available?(session) pid = pid_for(session.id) clear_pid(session.id) if pid.present? && !process_alive?(pid.to_i) @@ -66,42 +67,111 @@ module Streams last_restart = redis.get(restart_debounce_key(session.id)).to_i return if last_restart.positive? && (Time.now.to_i - last_restart) < 5 - start(session) + start_on_worker!(session) redis.set(restart_debounce_key(session.id), Time.now.to_i, ex: 300) rescue Error => e - Rails.logger.warn("[YoutubeRelay] ensure_publishing session=#{session.id}: #{e.message}") + Rails.logger.warn("[YoutubeRelay] ensure_on_worker session=#{session.id}: #{e.message}") + end + + def stop_on_worker!(session) + pid = pid_for(session.id) + return false if pid.blank? + + terminate_pid(pid) + clear_pid(session.id) + redis.del(format(OWNER_KEY, session.id)) + Rails.logger.info("[YoutubeRelay] stopped pid=#{pid} session=#{session.id}") + true end private - # Non usare ffprobe su RTMP live: si blocca e impedisce l'avvio del relay (Sidekiq in hang). - # Audio: traccia AAC da anullsrc (YouTube la richiede); il mic del telefono resta su _air/HLS. - def youtube_ffmpeg_args(intake, output) + def start_on_worker!(session) + return unless session.platform == "youtube" + return if session.stream_key.blank? + return if session.terminal? + return unless intake_available?(session) + + return pid_for(session.id).to_i if running?(session.id) + + stop_on_worker!(session) if pid_for(session.id).present? + + log_path = log_file(session) + FileUtils.mkdir_p(File.dirname(log_path)) + intake_source = mediamtx_intake_source(session) + output = "rtmps://a.rtmps.youtube.com/live2/#{session.stream_key}" + + pid = Process.spawn( + *youtube_ffmpeg_args(intake_source, output), + %i[out err] => log_path, + pgroup: true + ) + Process.detach(pid) + store_pid(session.id, pid) + redis.set(format(OWNER_KEY, session.id), worker_id, ex: 48.hours.to_i) + Rails.logger.info("[YoutubeRelay] started pid=#{pid} session=#{session.id} intake=#{intake_source.join(":")}") + schedule_youtube_activate(session) + pid + rescue Errno::ENOENT => e + raise Error, "ffmpeg non disponibile: #{e.message}" + end + + # _air RTMP (copy) se pronto; altrimenti HLS via proxy Rails. + def youtube_ffmpeg_args(intake_source, output) + mode, url = intake_source + if mode == :rtmp + return [ + "ffmpeg", "-nostdin", "-hide_banner", "-loglevel", "warning", + "-fflags", "+genpts+discardcorrupt", + "-analyzeduration", "10000000", "-probesize", "10000000", + "-rw_timeout", "15000000", + "-i", url, + "-c:v", "copy", + "-c:a", "aac", "-b:a", "128k", "-ar", "48000", "-ac", "1", + "-bsf:a", "aac_adtstoasc", + "-f", "flv", output + ] + end + [ "ffmpeg", "-nostdin", "-hide_banner", "-loglevel", "warning", - "-fflags", "+genpts+discardcorrupt", "-use_wallclock_as_timestamps", "1", - "-analyzeduration", "10000000", "-probesize", "10000000", + "-fflags", "+genpts+discardcorrupt", "-rw_timeout", "15000000", - "-i", intake, - "-f", "lavfi", "-i", "anullsrc=channel_layout=mono:sample_rate=48000", - "-map", "0:v:0", "-map", "1:a:0", - "-c:v", "libx264", "-preset", "veryfast", "-tune", "zerolatency", - "-pix_fmt", "yuv420p", "-r", "30", "-g", "30", "-keyint_min", "30", "-sc_threshold", "0", - "-b:v", "4000k", "-maxrate", "4000k", "-bufsize", "8000k", + "-live_start_index", "-1", + "-i", url, + "-c:v", "copy", "-c:a", "aac", "-b:a", "128k", "-ar", "48000", "-ac", "1", + "-bsf:a", "aac_adtstoasc", "-f", "flv", output ] end - def schedule_youtube_activate(session) - return if session.youtube_broadcast_id.blank? + def mediamtx_intake_source(session) + base = ENV.fetch("MEDIAMTX_INTERNAL_RTMP_URL", "rtmp://mediamtx:1935") + if Mediamtx::PublisherOnline.active?(session) + return [:rtmp, "#{base.chomp('/')}/#{session.mediamtx_path_name}"] + end - YoutubeBroadcastActivateJob.schedule_with_retries(session.id) + path = session.mediamtx_path_name + rails = ENV.fetch("RAILS_INTERNAL_URL", "http://rails:3000") + [:hls, "#{rails.chomp('/')}/hls/#{path}/index.m3u8"] end - def mediamtx_intake_url(session) - base = ENV.fetch("MEDIAMTX_INTERNAL_RTMP_URL", "rtmp://mediamtx:1935") - "#{base.chomp('/')}/#{session.mediamtx_overlay_path_name}" + def intake_available?(session) + return true if Mediamtx::PublisherOnline.active?(session) + + info = Mediamtx::Client.new.list_paths.find { |i| i["name"] == session.mediamtx_path_name } + info && (info["ready"] || info["online"] || info["available"]) + rescue StandardError + false + end + + def schedule_youtube_activate(session) + Youtube::LivePipeline.schedule_activate!(session, force: true) + end + + def worker_id + ENV.fetch("HOSTNAME", "worker") end def log_file(session) diff --git a/backend/app/services/webhooks/mediamtx_handler.rb b/backend/app/services/webhooks/mediamtx_handler.rb index dd3fe43..705aa95 100644 --- a/backend/app/services/webhooks/mediamtx_handler.rb +++ b/backend/app/services/webhooks/mediamtx_handler.rb @@ -25,8 +25,10 @@ module Webhooks def handle_connect(session) return if session.paused? - Streams::OverlayRelay.ensure_publishing!(session) - Streams::YoutubeRelay.ensure_publishing!(session) if session.platform == "youtube" + if Mediamtx::PublisherOnline.active?(session) && session.platform == "youtube" + Mediamtx::PublisherSync.schedule_youtube_pipeline!(session) + end + Streams::YoutubeRelay.stop(session) if session.platform == "youtube" return if session.live? && session.stream_events.where(event_type: "connected").exists? @@ -112,7 +114,6 @@ module Webhooks def broadcast(session, event) SessionChannel.broadcast_message(session, { type: "stream_event", event: event }) - Streams::Overlay::Refresh.call(session) end end end diff --git a/backend/app/services/youtube/api_throttle.rb b/backend/app/services/youtube/api_throttle.rb new file mode 100644 index 0000000..9ef5b9b --- /dev/null +++ b/backend/app/services/youtube/api_throttle.rb @@ -0,0 +1,28 @@ +module Youtube + # Evita tempeste di chiamate YouTube Data API (quota / userRequestsExceedRateLimit). + class ApiThrottle + KEY = "youtube_api:rate_limited_until" + + class << self + def rate_limited? + until_ts = redis.get(KEY).to_i + until_ts.positive? && until_ts > Time.now.to_i + end + + def mark_rate_limited!(seconds: 120) + redis.set(KEY, (Time.now.to_i + seconds), ex: seconds + 30) + end + + def rate_limit_error?(message) + msg = message.to_s + msg.include?("userRequestsExceedRateLimit") || msg.include?("rate limit") + end + + private + + def redis + @redis ||= Redis.new(url: ENV.fetch("REDIS_URL", "redis://localhost:6379/0")) + end + end + end +end diff --git a/backend/app/services/youtube/broadcast_service.rb b/backend/app/services/youtube/broadcast_service.rb index 3cb9d73..f731dc1 100644 --- a/backend/app/services/youtube/broadcast_service.rb +++ b/backend/app/services/youtube/broadcast_service.rb @@ -47,14 +47,16 @@ module Youtube rtmp_url: stream_result.cdn.ingestion_info.ingestion_address } rescue Google::Apis::Error => e + ApiThrottle.mark_rate_limited! if ApiThrottle.rate_limit_error?(e.message) raise Error, friendly_api_error(e) end # Porta in onda appena c’è ingest su YouTube (non attendere scheduledStartTime). - # @return [Symbol] :live, :waiting_ingest, :transitioned, :skipped + # @return [Symbol] :live, :waiting_ingest, :transitioned, :skipped, :rate_limited def activate_broadcast!(broadcast_id) return :skipped if broadcast_id.blank? return :skipped if @credential.blank? || missing_oauth_config? + return :rate_limited if ApiThrottle.rate_limited? client = authorized_client item = client.list_live_broadcasts("status,contentDetails", id: broadcast_id).items&.first @@ -68,7 +70,11 @@ module Youtube if stream_id.present? stream = client.list_live_streams("status", id: stream_id).items&.first stream_status = stream&.status&.stream_status.to_s + Rails.logger.info("[Youtube::BroadcastService] activate #{broadcast_id}: lifecycle=#{status} stream_status=#{stream_status.presence || "(blank)"}") return :waiting_ingest unless stream_status == "active" + else + Rails.logger.info("[Youtube::BroadcastService] activate #{broadcast_id}: lifecycle=#{status} stream_status=(unbound)") + return :waiting_ingest end if %w[ready created].include?(status) @@ -82,6 +88,11 @@ module Youtube :skipped rescue Google::Apis::Error => e + if ApiThrottle.rate_limit_error?(e.message) + ApiThrottle.mark_rate_limited! + Rails.logger.warn("[Youtube::BroadcastService] activate rate limited #{broadcast_id}") + return :rate_limited + end Rails.logger.warn("[Youtube::BroadcastService] activate #{broadcast_id}: #{e.message}") :skipped end @@ -133,6 +144,8 @@ module Youtube return "Orario programmazione YouTube non valido. Riprova: la diretta partirà appena invii il segnale RTMP." end + return "YouTube temporaneamente occupato. Riprova tra 1–2 minuti." if ApiThrottle.rate_limit_error?(msg) + msg end diff --git a/backend/app/services/youtube/live_pipeline.rb b/backend/app/services/youtube/live_pipeline.rb new file mode 100644 index 0000000..570f463 --- /dev/null +++ b/backend/app/services/youtube/live_pipeline.rb @@ -0,0 +1,107 @@ +module Youtube + # Broadcast YouTube (API) + relay ffmpeg copy da MediaMTX → RTMPS. Nessun overlay. + class LivePipeline + PIPELINE_DEBOUNCE_KEY = "youtube_pipeline:debounce:%s" + ACTIVATE_DEBOUNCE_KEY = "youtube_activate:debounce:%s" + SETUP_DEBOUNCE_KEY = "youtube_setup:debounce:%s" + PUBLISHER_SEEN_KEY = "publisher_online:seen:%s" + + class << self + def schedule!(session, force: false) + return unless pipeline_eligible?(session) + + ensure_broadcast_setup!(session) unless broadcast_ready?(session) + + return unless force || redis.set(format(PIPELINE_DEBOUNCE_KEY, session.id), "1", nx: true, ex: 8) + + Streams::YoutubeRelay.ensure_publishing!(session) if broadcast_ready?(session) + if broadcast_ready?(session) + YoutubeIngestWatchdogJob.schedule_chain(session.id) + schedule_activate!(session, force: true, wait: 20.seconds) + end + Rails.logger.info("[Youtube::LivePipeline] scheduled session=#{session.id} broadcast_ready=#{broadcast_ready?(session)}") + end + + def schedule_activate!(session, force: false, wait: 8.seconds) + return unless pipeline_eligible?(session) + return unless broadcast_ready?(session) + + debounce_ok = force || redis.set(format(ACTIVATE_DEBOUNCE_KEY, session.id), "1", nx: true, ex: 12) + return unless debounce_ok + + YoutubeBroadcastActivateJob.set(wait: wait).perform_later(session.id, 1) + end + + def ensure_broadcast_setup!(session) + return unless pipeline_eligible?(session) + return if broadcast_ready?(session) + return if redis.exists?(format(YoutubeBroadcastSetupJob::RUN_LOCK_KEY, session.id)).positive? + + return unless redis.set(format(SETUP_DEBOUNCE_KEY, session.id), "1", nx: true, ex: 90) + + YoutubeBroadcastSetupJob.perform_later(session.id) + Rails.logger.info("[Youtube::LivePipeline] queued YoutubeBroadcastSetupJob session=#{session.id}") + end + + def tick_session!(session) + return unless pipeline_eligible?(session) + + ensure_broadcast_setup!(session) unless broadcast_ready?(session) + + online = Mediamtx::PublisherOnline.active?(session) + seen_key = format(PUBLISHER_SEEN_KEY, session.id) + was_online = redis.get(seen_key) == "1" + redis.set(seen_key, online ? "1" : "0", ex: 4.hours.to_i) + + unless online + Streams::YoutubeRelay.ensure_publishing!(session) if broadcast_ready?(session) + return + end + + if session.paused? + Streams::YoutubeRelay.ensure_publishing!(session) if broadcast_ready?(session) + return + end + + Mediamtx::Client.new.set_always_available(session, enabled: false) + session.go_live! if session.may_go_live? + session.reconnect! if session.reconnecting? && session.may_reconnect? + + if !was_online + Rails.logger.info("[Youtube::LivePipeline] publisher online session=#{session.id}") + schedule!(session, force: true) + return + end + + if Streams::YoutubeRelay.running?(session.id) + schedule_activate!(session) + else + schedule!(session) + end + rescue Mediamtx::Client::Error => e + Rails.logger.warn("[Youtube::LivePipeline] tick session=#{session.id}: #{e.message}") + end + + def pipeline_eligible?(session) + session.present? && + session.platform == "youtube" && + !session.terminal? && + session.status.in?(%w[live connecting reconnecting paused]) + end + + def broadcast_ready?(session) + session.youtube_broadcast_id.present? && session.stream_key.present? + end + + def eligible?(session) + pipeline_eligible?(session) && broadcast_ready?(session) + end + + private + + def redis + @redis ||= Redis.new(url: ENV.fetch("REDIS_URL", "redis://localhost:6379/0")) + end + end + end +end diff --git a/backend/app/services/youtube/setup_broadcast.rb b/backend/app/services/youtube/setup_broadcast.rb index 9dbf582..7585cf3 100644 --- a/backend/app/services/youtube/setup_broadcast.rb +++ b/backend/app/services/youtube/setup_broadcast.rb @@ -36,7 +36,7 @@ module Youtube @session.reload return @session if @session.terminal? return @session if @session.platform != "youtube" - return @session if @session.youtube_broadcast_id.present? + return @session if @session.youtube_broadcast_id.present? && @session.stream_key.present? yt = BroadcastService.new(@session.match.team, youtube_channel: @youtube_channel) title = "#{@session.match.team.name} vs #{@session.match.opponent_name}" @@ -53,10 +53,7 @@ module Youtube rtmp_url: broadcast[:rtmp_url] ) - # Relay solo quando _air pubblica (overlay pronto); altrimenti ffmpeg esce subito. - if Streams::OverlayRelay.running?(@session.id) - Streams::YoutubeRelay.start(@session) - end + # Relay copy MediaMTX → YouTube parte subito (copertina slate se il telefono non c'è ancora). @session.stream_events.create!( event_type: "youtube_ready", metadata: { broadcast_id: broadcast[:broadcast_id] }, @@ -71,6 +68,7 @@ module Youtube youtube_broadcast_id: @session.youtube_broadcast_id } ) + Youtube::LivePipeline.schedule!(@session, force: true) @session rescue BroadcastService::Error => e @session.stream_events.create!( diff --git a/backend/app/views/public/live/index.html.erb b/backend/app/views/public/live/index.html.erb index e09d93c..428cd3a 100644 --- a/backend/app/views/public/live/index.html.erb +++ b/backend/app/views/public/live/index.html.erb @@ -61,7 +61,7 @@
<% @sessions.each do |session| %> <% match = session.match %> - <% on_air = @online_paths.include?(session.mediamtx_path_name) %> + <% on_air = @online_paths.include?(session.mediamtx_overlay_path_name) || @online_paths.include?(session.mediamtx_path_name) %>
<%= live_match_card_heading(match) %>

@@ -82,6 +82,8 @@ In onda <% elsif session.status == "live" %> Live + <% elsif session.status == "paused" %> + In pausa <% elsif session.status == "reconnecting" %> Riconnessione <% else %> diff --git a/backend/app/views/public/live/show.html.erb b/backend/app/views/public/live/show.html.erb index 554b852..354627f 100644 --- a/backend/app/views/public/live/show.html.erb +++ b/backend/app/views/public/live/show.html.erb @@ -40,7 +40,7 @@

<% else %>
- + <%# Tabellone, badge stato e brand sono bruciati nel video (Streams::OverlayRelay). %>