MatchLiveTv/backend/app/services/streams/youtube_relay.rb

module Streams
  # 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 worker?
        start_on_worker!(session)
      end

      def stop(session)
        clear_pid(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)
        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

      def ensure_publishing!(session)
        return unless session.platform == "youtube"
        return if session.terminal?
        return if session.stream_key.blank?

        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)

        return if running?(session.id)

        last_restart = redis.get(restart_debounce_key(session.id)).to_i
        return if last_restart.positive? && (Time.now.to_i - last_restart) < 5

        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_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

      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

      # RTMP grezzo da telefono (overlay bruciato lato app); fallback 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",
          "-rw_timeout", "15000000",
          "-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 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

        path = session.mediamtx_path_name
        hls = ENV.fetch("MEDIAMTX_HLS_URL", "http://mediamtx:8888").chomp("/")
        [:hls, "#{hls}/#{path}/index.m3u8"]
      end

      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)
        Rails.root.join("log", "youtube_relay_#{session.id}.log").to_s
      end

      def redis
        @redis ||= Redis.new(url: ENV.fetch("REDIS_URL", "redis://localhost:6379/0"))
      end

      def restart_debounce_key(session_id)
        format("youtube_relay:debounce:%s", session_id)
      end

      def store_pid(session_id, pid)
        redis.set(format(REDIS_KEY, session_id), pid, ex: 48.hours.to_i)
      end

      def clear_pid(session_id)
        redis.del(format(REDIS_KEY, session_id))
      end

      def pid_for(session_id)
        redis.get(format(REDIS_KEY, session_id))
      end

      def process_alive?(pid)
        stat = File.read("/proc/#{pid.to_i}/stat")
        return false if stat.split[2] == "Z"

        Process.kill(0, pid.to_i)
        true
      rescue Errno::ESRCH, Errno::EPERM, Errno::ENOENT
        false
      end

      def terminate_pid(pid)
        Process.kill("TERM", -pid.to_i)
        sleep 0.5
      rescue Errno::ESRCH
        nil
      else
        begin
          Process.kill("KILL", -pid.to_i)
        rescue Errno::ESRCH
          nil
        end
      end
    end
  end
end