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 RECORDING_PATCH_COOLDOWN = -> { ENV.fetch("MEDIAMTX_RECORDING_PATCH_COOLDOWN_SECS", "45").to_i.seconds } def initialize(session) @session = session end def call return @session if @session.terminal? path_info = Mediamtx::PublisherOnline.path_info(@session) publisher_online = Mediamtx::PublisherOnline.active_path?(path_info) if publisher_online clear_publisher_misses!(@session.id) if @session.paused? # RTMP ancora connesso in pausa: non forzare live/reconnect. else enable_live_path_once!(@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) disable_recording_on_path!(@session, path_info: path_info) restore_slate_path!(@session) @session.lose_connection! schedule_timeout unless @session.paused? end elsif !publisher_online && @session.status.in?(%w[reconnecting connecting]) && !@session.paused? # Non togliere recording qui: un gap RTMP breve non deve ricaricare la config MediaMTX. restore_slate_path!(@session) end desired_recording = recording_should_be_active?(publisher_online) sync_path_recording!(@session, enabled: desired_recording, path_info: path_info) @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 def enable_live_path_once!(session) return unless session.platform == "youtube" key = format("youtube:slate_disabled:%s", session.id) return unless redis.set(key, "1", nx: true, ex: 48.hours.to_i) Client.new.set_always_available(session, enabled: false) rescue Client::Error => e redis.del(format("youtube:slate_disabled:%s", session.id)) 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 def recording_should_be_active?(publisher_online) return false unless @session.match.team.entitlements.recording_enabled_for_mediamtx? return false if @session.paused? || @session.terminal? return false unless publisher_online return false unless @session.status.in?(%w[live reconnecting]) true end def disable_recording_on_path!(session, path_info:) sync_path_recording!(session, enabled: false, path_info: path_info, force: true) end # In produzione i webhook MediaMTX (wget) non girano: la registrazione va accesa al primo go-live. # Evitiamo patch ripetute: ogni reload config distrugge i muxer HLS (500 sul proxy /hls/). def sync_path_recording!(session, enabled:, path_info: nil, force: false) 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 return if path_info && path_recording_active?(path_info) == enabled if !enabled && !force && recording_patch_on_cooldown?(session.id) Rails.logger.debug("[PublisherSync] skip recording patch session=#{session.id} (cooldown)") return end Client.new.set_path_recording(session, enabled: enabled) redis.set(key, desired, ex: 48.hours.to_i) mark_recording_patch!(session.id) rescue Client::Error => e Rails.logger.warn("[PublisherSync] recording enabled=#{enabled} session=#{session.id}: #{e.message}") end def path_recording_active?(path_info) path_info["record"] == true end def recording_patch_on_cooldown?(session_id) last = redis.get(recording_patch_at_key(session_id)).to_i last.positive? && Time.at(last) > RECORDING_PATCH_COOLDOWN.call.ago end def mark_recording_patch!(session_id) redis.set(recording_patch_at_key(session_id), Time.current.to_i, ex: 48.hours.to_i) end def recording_patch_at_key(session_id) format("mediamtx:recording:patched_at:%s", session_id) 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}") redis.del(format("mediamtx:recording:patched_at:%s", 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