require "set" module Mediamtx class Client class Error < StandardError; end def self.for_session(session) new(base_url: session.mediamtx_api_base_url) end def initialize(base_url: MatchLiveTv.mediamtx_api_url) @base_url = base_url @conn = Faraday.new(url: base_url) do |f| f.request :json f.response :json f.adapter Faraday.default_adapter end end attr_reader :base_url def create_path(session) path = session.mediamtx_path_name # record: false finché non c'è publisher — con alwaysAvailable MediaMTX registrerebbe # solo la slate in pausa/attesa. Slate resta accesa anche su YouTube (copertina se l'app cade). body = recording_body(session, enabled: false).merge( source: "publisher", overridePublisher: true ) body[:alwaysAvailable] = true body[:alwaysAvailableFile] = slate_file_path(session) # 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) return true if path_name.blank? response = @conn.delete("/v3/config/paths/delete/#{CGI.escape(path_name)}") return true if response.status == 404 unless response.success? err = response.body.is_a?(Hash) ? response.body["error"] : response.body raise Error, "MediaMTX path delete failed: #{response.status} #{err}" end forget_always_available(path_name) true end def set_path_recording(session, enabled:) path = session.mediamtx_path_name body = recording_body(session, enabled: enabled) 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 recording patch failed: #{response.status} #{err}" end true end def patch_path(session) set_always_available(session, enabled: true) end # Garantisce path config + slate: se MediaMTX ha perso la config (restart → all_others), # in pausa non resterebbe nessuno stream verso HLS/YouTube. def ensure_path_with_slate!(session) path = session.mediamtx_path_name response = @conn.get("/v3/config/paths/get/#{CGI.escape(path)}") missing = response.status == 404 || (response.body.is_a?(Hash) && response.body["error"].present?) if missing forget_always_available(path) begin create_path(session) rescue Error # Path creato in parallelo o già presente: forza solo la slate. forget_always_available(path) set_always_available(session, enabled: true) end return true end conf = response.body.is_a?(Hash) ? response.body : {} desired = slate_file_path(session) if conf["alwaysAvailable"] == true && conf["alwaysAvailableFile"].to_s == desired.to_s remember_always_available(path, enabled: true) return true end forget_always_available(path) set_always_available(session, enabled: true) end # Spegnere record solo se attivo: ogni PATCH path ricarica MediaMTX e distrugge i muxer HLS. def disable_recording_if_active!(session) path = session.mediamtx_path_name response = @conn.get("/v3/config/paths/get/#{CGI.escape(path)}") return true unless response.success? return true unless response.body.is_a?(Hash) && response.body["record"] == true set_path_recording(session, enabled: false) end # Slate alwaysAvailable: copertina sullo stesso path quando il telefono è offline. # Resta accesa anche con publisher in onda (MediaMTX usa il publisher se presente). def set_always_available(session, enabled:) path = session.mediamtx_path_name return true if always_available_remembered?(path, enabled: enabled) body = if enabled { alwaysAvailable: true, alwaysAvailableFile: slate_file_path(session) } 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 alwaysAvailable patch failed: #{response.status} #{err}" end remember_always_available(path, enabled: enabled) true end def list_paths response = @conn.get("/v3/paths/list") return [] unless response.success? body = response.body body.is_a?(Hash) ? (body["items"] || []) : [] rescue Error, Faraday::Error [] end def list_rtmp_conns response = @conn.get("/v3/rtmpconns/list") return [] unless response.success? body = response.body body.is_a?(Hash) ? (body["items"] || []) : [] rescue Error, Faraday::Error [] end def online_path_names Set.new(list_paths.filter_map { |item| item["name"] if item["online"] }) end private def recording_body(session, enabled:) ent = session.match.team.entitlements can_record = ent.recording_enabled_for_mediamtx? body = { record: enabled && can_record, recordPath: "/recordings/%path/%Y-%m-%d_%H-%M-%S", recordSegmentDuration: "60s" } if enabled && can_record retention_hours = ent.recording_retention_days * 24 + 48 body[:recordDeleteAfter] = "#{retention_hours}h" end body end def publish_webhook(session) webhook_curl(session, "connect") end def disconnect_webhook(session) webhook_curl(session, "disconnect") end def webhook_curl(session, event) secret = MatchLiveTv.mediamtx_webhook_secret rails = ENV.fetch("RAILS_WEBHOOK_URL", "http://rails:3000") payload = %({"session_id":"#{session.id}"}) # MediaMTX image non include curl; wget è disponibile nell'immagine ufficiale <<~SCRIPT.squish wget -q -O- --post-data='#{payload}' --header='Content-Type: application/json' --header="X-MediaMTX-Signature: $(printf '%s' '#{payload}' | openssl dgst -sha256 -hmac '#{secret}' | cut -d' ' -f2)" #{rails}/webhooks/mediamtx/#{event} SCRIPT end def run_on_ready_script(session) connect = webhook_curl(session, "connect") return connect if session.matchlivetv_platform? "#{connect} & #{relay_script(session)}" end def relay_script(session) return "echo 'youtube relay skipped'" unless session.platform == "youtube" key = session.stream_key path = session.mediamtx_path_name <<~SCRIPT.squish ffmpeg -re -i rtmp://127.0.0.1:1935/#{path} -c copy -f flv rtmp://a.rtmp.youtube.com/live2/#{key} 2>/var/log/ffmpeg-#{path}.log SCRIPT end def slate_file_path(session) return default_slate_file_path if session.blank? Streams::SlateDistributor.slate_path_for_session(session) end def default_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