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

module Streams
  # Transcodifica path grezzo → path _air con overlay PNG (tabellone + badge) bruciato nel video.
  class OverlayRelay
    class Error < StandardError; end

    REDIS_KEY = "overlay_relay:pid:%s"

    class << self
      def start(session)
        return if session.terminal?

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

        Streams::Overlay::Refresh.call(session)

        dir = Streams::Overlay.overlay_dir(session.id)
        FileUtils.mkdir_p(dir)
        png = Streams::Overlay.overlay_png_path(session)
        pipe = Streams::Overlay.overlay_pipe_path(session)
        ensure_pipe!(pipe)
        intake = "#{mediamtx_rtmp_url}/#{session.mediamtx_path_name}"
        output = "#{mediamtx_rtmp_url}/#{session.mediamtx_overlay_path_name}"
        log_path = log_file(session)
        FileUtils.mkdir_p(File.dirname(log_path))

        fps = output_fps
        overlay_fps = overlay_input_fps
        filter = "[1:v]format=rgba[ol];[0:v]fps=#{fps},format=yuv420p[vid];[vid][ol]overlay=0:0:format=auto[vout]"

        ffmpeg_pid = Process.spawn(
          "ffmpeg", "-nostdin", "-hide_banner", "-loglevel", "warning",
          "-fflags", "+genpts+discardcorrupt",
          "-analyzeduration", "10000000", "-probesize", "10000000",
          "-rw_timeout", "15000000",
          "-i", intake,
          "-f", "image2pipe", "-framerate", overlay_fps.to_s, "-i", pipe.to_s,
          "-filter_complex", filter,
          "-map", "[vout]", "-map", "0:a?",
          "-fps_mode", "cfr", "-r", fps.to_s,
          "-c:v", "libx264", "-preset", x264_preset, "-bf", "0",
          "-profile:v", "high", "-pix_fmt", "yuv420p",
          "-b:v", bitrate_for(session), "-maxrate", maxrate_for(session), "-bufsize", bufsize_for(session),
          "-g", fps.to_s, "-keyint_min", fps.to_s,
          "-c:a", "aac", "-b:a", "128k", "-ar", "48000", "-ac", "1",
          "-f", "flv", output,
          %i[out err] => log_path,
          pgroup: true
        )
        spawn_png_feeder!(ffmpeg_pid, pipe, png, log_path)
        Process.detach(ffmpeg_pid)
        store_pid(session.id, ffmpeg_pid)
        OverlayRefreshJob.schedule_chain(session.id)
        Rails.logger.info("[OverlayRelay] started pid=#{ffmpeg_pid} session=#{session.id}")
        ffmpeg_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)
        pipe = Streams::Overlay.overlay_pipe_path(session)
        FileUtils.rm_f(pipe) if pipe.exist?
        OverlayRefreshJob.cancel_chain(session.id)
        Rails.logger.info("[OverlayRelay] stopped pid=#{pid} session=#{session.id}")
        true
      end

      def running?(session_id)
        pid = pid_for(session_id)
        return false if pid.blank?

        return true unless overlay_relay_process_check_local?

        process_alive?(pid)
      end

      # Riavvia il relay se il processo è morto o _air non pubblica (es. feeder bloccato).
      def ensure_publishing!(session)
        return if session.terminal?

        if running?(session.id) && overlay_path_publishing?(session)
          return
        end

        if running?(session.id)
          misses = redis.incr(overlay_air_miss_key(session.id)).to_i
          redis.expire(overlay_air_miss_key(session.id), 120)
          return if misses < 3

          redis.del(overlay_air_miss_key(session.id))
          Rails.logger.warn("[OverlayRelay] _air non attivo dopo #{misses} tentativi, restart session=#{session.id}")
          stop(session)
        else
          redis.del(overlay_air_miss_key(session.id))
        end

        start(session)
      rescue Error => e
        Rails.logger.warn("[OverlayRelay] ensure_publishing session=#{session.id}: #{e.message}")
      end

      private

      def mediamtx_rtmp_url
        ENV.fetch("MEDIAMTX_INTERNAL_RTMP_URL", "rtmp://mediamtx:1935").chomp("/")
      end

      # Ricodifica con overlay: serve più budget della sorgente (generational loss).
      def output_video_kbps(session)
        override = ENV["OVERLAY_RELAY_VIDEO_KBPS"].presence&.to_i
        return override if override&.positive?

        source_kbps = [(session.target_bitrate.to_i / 1000), 2500].max
        [(source_kbps * 1.8).to_i, 4500].max
      end

      def bitrate_for(session)
        "#{output_video_kbps(session)}k"
      end

      def maxrate_for(session)
        kbps = (output_video_kbps(session) * 1.15).to_i
        "#{kbps}k"
      end

      def bufsize_for(session)
        kbps = output_video_kbps(session) * 2
        "#{kbps}k"
      end

      def output_fps
        ENV.fetch("OVERLAY_RELAY_FPS", "30").to_i
      end

      def x264_preset
        ENV.fetch("OVERLAY_RELAY_X264_PRESET", "fast")
      end

      def overlay_input_fps
        ENV.fetch("OVERLAY_INPUT_FPS", "2").to_i
      end

      def ensure_pipe!(path)
        FileUtils.rm_f(path) if path.exist? && !path.pipe?
        system("mkfifo", path.to_s) unless path.exist?
        raise Error, "impossibile creare fifo overlay #{path}" unless path.pipe?
      end

      def spawn_png_feeder!(ffmpeg_pid, pipe, png, log_path)
        feeder = Rails.root.join("bin/overlay_png_feeder.rb")
        poll = (1.0 / overlay_input_fps).round(2)
        Process.spawn(
          RbConfig.ruby, feeder.to_s, pipe.to_s, png.to_s, poll.to_s,
          %i[out err] => log_path,
          pgroup: ffmpeg_pid
        )
      end

      def overlay_relay_process_check_local?
        ENV["OVERLAY_RELAY_PROCESS_CHECK"] != "false"
      end

      def overlay_path_publishing?(session)
        info = Mediamtx::Client.new.list_paths.find { |i| i["name"] == session.mediamtx_overlay_path_name }
        info && (info["online"] || info["ready"] || info["available"])
      end

      def log_file(session)
        Rails.root.join("log", "overlay_relay_#{session.id}.log").to_s
      end

      def redis
        @redis ||= Redis.new(url: ENV.fetch("REDIS_URL", "redis://localhost:6379/0"))
      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 overlay_air_miss_key(session_id)
        format("overlay_relay:air_miss:%s", 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