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

module Streams
  # Relay continuo verso YouTube: legge l'uscita MediaMTX del path (camera o slate alwaysAvailable)
  # senza interrompere FLV quando il telefono si mette in pausa o perde rete.
  class YoutubeRelay
    class Error < StandardError; end

    REDIS_KEY = "youtube_relay:pid:%s"

    class << self
      def start(session)
        return unless session.platform == "youtube"
        return if session.stream_key.blank?
        return if session.terminal?

        stop(session) if pid_for(session.id).present?

        log_path = log_file(session)
        FileUtils.mkdir_p(File.dirname(log_path))
        intake = mediamtx_intake_url(session)
        output = "rtmp://a.rtmp.youtube.com/live2/#{session.stream_key}"

        # YouTube richiede FLV H.264/AAC stabili; -c copy lascia spesso streamStatus=inactive.
        pid = Process.spawn(
          *youtube_ffmpeg_args(intake, output, log_path),
          pgroup: true
        )
        Process.detach(pid)
        store_pid(session.id, pid)
        Rails.logger.info("[YoutubeRelay] started pid=#{pid} session=#{session.id}")
        schedule_youtube_activate(session)
        pid
      rescue Errno::ENOENT => e
        raise Error, "ffmpeg non disponibile: #{e.message}"
      end

      def stop(session)
        pid = pid_for(session.id)
        return false if pid.blank?

        terminate_pid(pid)
        clear_pid(session.id)
        Rails.logger.info("[YoutubeRelay] stopped pid=#{pid} session=#{session.id}")
        true
      end

      def running?(session_id)
        pid = pid_for(session_id)
        pid.present? && process_alive?(pid)
      end

      # Riavvia il relay se è morto (es. avviato prima del segnale su _air).
      def ensure_publishing!(session)
        return unless session.platform == "youtube"
        return if session.terminal?
        return if session.stream_key.blank?
        return unless Streams::OverlayRelay.running?(session.id)

        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) < 20

        start(session)
        redis.set(restart_debounce_key(session.id), Time.now.to_i, ex: 300)
      rescue Error => e
        Rails.logger.warn("[YoutubeRelay] ensure_publishing session=#{session.id}: #{e.message}")
      end

      private

      def youtube_ffmpeg_args(intake, output, log_path)
        [
          "ffmpeg", "-nostdin", "-hide_banner", "-loglevel", "warning",
          "-fflags", "+genpts+discardcorrupt", "-use_wallclock_as_timestamps", "1",
          "-rw_timeout", "15000000",
          "-i", intake,
          "-map", "0:v:0", "-map", "0:a:0",
          "-c:v", "libx264", "-preset", "veryfast", "-tune", "zerolatency",
          "-pix_fmt", "yuv420p", "-r", "30", "-g", "30", "-keyint_min", "30", "-sc_threshold", "0",
          "-b:v", "4000k", "-maxrate", "4000k", "-bufsize", "8000k",
          "-c:a", "aac", "-b:a", "128k", "-ar", "48000", "-ac", "1",
          "-f", "flv", output,
          %i[out err] => log_path
        ]
      end

      def schedule_youtube_activate(session)
        return if session.youtube_broadcast_id.blank?

        YoutubeBroadcastActivateJob.set(wait: 8.seconds).perform_later(session.id)
      end

      def mediamtx_intake_url(session)
        base = ENV.fetch("MEDIAMTX_INTERNAL_RTMP_URL", "rtmp://mediamtx:1935")
        "#{base.chomp('/')}/#{session.mediamtx_overlay_path_name}"
      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)
        Process.kill(0, pid.to_i)
        true
      rescue Errno::ESRCH, Errno::EPERM
        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