module Mediamtx # MediaMTX in produzione è distroless: runOnReady con wget non funziona. # Sincronizziamo lo stato Rails leggendo l'API paths (poll da /live/:id/status.json). class PublisherSync def initialize(session) @session = session end def call return @session if @session.terminal? info = Mediamtx::Client.new.list_paths.find { |i| i["name"] == @session.mediamtx_path_name } 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. sync_path_recording!(@session, enabled: false) else sync_path_recording!(@session, enabled: true) 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) if @session.platform == "youtube" end kick_youtube!(@session) if @session.platform == "youtube" end elsif !@session.paused? && @session.live? && @session.may_lose_connection? if publisher_offline_sustained?(@session.id) clear_publisher_misses!(@session.id) sync_path_recording!(@session, enabled: false) restore_slate_path!(@session) @session.lose_connection! schedule_timeout unless @session.paused? end elsif !publisher_online && @session.status.in?(%w[reconnecting connecting]) && !@session.paused? sync_path_recording!(@session, enabled: false) restore_slate_path!(@session) end sync_path_recording!(@session, enabled: false) if @session.paused? && !publisher_online @session.reload.tap do |session| schedule_youtube_relay_if_needed(session) end end private def kick_youtube!(session) return unless Youtube::LivePipeline.pipeline_eligible?(session) Youtube::LivePipeline.tick_session!(session) end def cancel_timeout return if @session.timeout_job_id.blank? Sidekiq::ScheduledSet.new.find_job(@session.timeout_job_id)&.delete @session.update!(timeout_job_id: nil) end def schedule_timeout job = DisconnectionTimeoutJob.perform_in( MatchLiveTv.reconnect_timeout_seconds.seconds, @session.id ) @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 # In produzione i webhook MediaMTX (wget) non girano: la registrazione va accesa qui. def sync_path_recording!(session, enabled:) return unless session.match.team.entitlements.recording_enabled_for_mediamtx? key = recording_state_key(session.id) desired = enabled ? "1" : "0" return if redis.get(key) == desired Client.new.set_path_recording(session, enabled: enabled) redis.set(key, desired, ex: 48.hours.to_i) rescue Client::Error => e Rails.logger.warn("[PublisherSync] recording enabled=#{enabled} session=#{session.id}: #{e.message}") end def self.forget_recording_state!(session_id) redis = Redis.new(url: ENV.fetch("REDIS_URL", "redis://localhost:6379/0")) redis.del("mediamtx:recording:#{session_id}") rescue Redis::BaseError nil end def recording_state_key(session_id) "mediamtx:recording:#{session_id}" 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