Aggiunge overlay server-side burn-in e stabilizza diretta live.

Tabellone, badge e brand sono ricodificati nel flusso _air via OverlayRelay;
sync punteggio HTTP, volume condiviso rails/sidekiq, qualità 720p migliorata
e badge CONNECTING in ripresa da pausa.

Co-authored-by: Cursor <cursoragent@cursor.com>

backend/app/services/streams/overlay_relay.rb

@@ -0,0 +1,205 @@
+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",
+          "-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)
+          Rails.logger.warn("[OverlayRelay] _air non attivo, restart session=#{session.id}")
+          stop(session)
+        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 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