MatchLiveTv/backend/app/services/mediamtx/client.rb

require "set"

module Mediamtx
  class Client
    class Error < StandardError; end

    CREATE_PATH_RETRIES = -> { ENV.fetch("MEDIAMTX_CREATE_RETRIES", "5").to_i }
    CREATE_PATH_RETRY_BASE_SECS = -> { ENV.fetch("MEDIAMTX_CREATE_RETRY_BASE_SECS", "0.4").to_f }

    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

    # Health probe for CPX readiness (GET /v3/paths/list).
    def reachable?(timeout: 2)
      conn = Faraday.new(url: @base_url) do |f|
        f.adapter Faraday.default_adapter
        f.options.open_timeout = timeout
        f.options.timeout = timeout
      end
      response = conn.get("/v3/paths/list")
      response.success?
    rescue Faraday::Error
      false
    end

    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 = with_connection_retries("create_path #{path}") do
        @conn.post("/v3/config/paths/add/#{CGI.escape(path)}", body)
      end
      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 with_connection_retries(label)
      attempts = [CREATE_PATH_RETRIES.call, 1].max
      base = CREATE_PATH_RETRY_BASE_SECS.call
      try = 0
      begin
        try += 1
        yield
      rescue Faraday::ConnectionFailed, Faraday::TimeoutError => e
        raise if try >= attempts

        sleep_secs = base * (2**(try - 1))
        Rails.logger.warn("[Mediamtx::Client] #{label} retry #{try}/#{attempts} after #{e.class}: #{e.message} (sleep #{sleep_secs}s)")
        sleep(sleep_secs)
        retry
      end
    end

    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