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"
    LIVE_INTAKE_KEY = "overlay_relay:live_intake:%s"
    MODE_KEY = "overlay_relay:mode:%s"
    START_LOCK_KEY = "overlay_relay:start_lock:%s"
    MODE_LIVE = "live".freeze
    MODE_COVER = "cover".freeze

    class << self
      def start(session, mode: nil)
        return if session.terminal?
        return unless acquire_start_lock!(session.id)

        begin
          terminate_all_session_processes!(session.id)

          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)
          log_path = log_file(session)
          FileUtils.mkdir_p(File.dirname(log_path))

          mode = normalize_mode(mode || desired_mode(session))
          ffmpeg_pid = Process.spawn(*ffmpeg_command(session, pipe, mode), %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)
          store_mode(session.id, mode)
          mode == MODE_LIVE ? mark_live_intake!(session.id) : clear_live_intake!(session.id)
          OverlayRefreshJob.schedule_chain(session.id)
          if session.platform == "youtube" && session.stream_key.present?
            Streams::YoutubeRelay.stop(session)
            Youtube::LivePipeline.schedule_activate!(session, force: true, wait: 20.seconds)
          end
          Rails.logger.info("[OverlayRelay] started pid=#{ffmpeg_pid} mode=#{mode} session=#{session.id} youtube_tee=#{session.platform == "youtube"}")
          ffmpeg_pid
        ensure
          release_start_lock!(session.id)
        end
      rescue Errno::ENOENT => e
        raise Error, "ffmpeg non disponibile: #{e.message}"
      end

      def stop(session)
        pids = session_process_pids(session.id)
        redis_pid = pid_for(session.id).to_i
        pids << redis_pid if redis_pid > 1
        pids = pids.uniq.select { |p| p > 1 }
        return false if pids.empty?

        pids.each { |p| terminate_pid(p) }
        clear_pid(session.id)
        clear_mode(session.id)
        clear_live_intake!(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 pids=#{pids.join(",")} session=#{session.id}")
        true
      end

      def running?(session_id)
        return false unless overlay_relay_process_check_local?

        alive = session_process_pids(session_id).select { |p| process_alive?(p) }
        return true if alive.any?

        redis_pid = pid_for(session_id).to_i
        return false if redis_pid <= 1

        process_alive?(redis_pid)
      end

      # Overlay ffmpeg vivo e path _air riceve dati.
      def healthy?(session)
        running?(session.id) && overlay_path_publishing?(session)
      end

      # Riavvia il relay se il processo è morto o _air non pubblica (es. feeder bloccato).
      def ensure_publishing!(session)
        return if session.terminal?
        restart_if_missing_youtube_tee!(session)
        wanted_mode = desired_mode(session)
        pids = session_process_pids(session.id)
        if pids.any? && pids.none? { |p| process_alive?(p) }
          terminate_all_session_processes!(session.id)
        end
        if healthy?(session) && current_mode(session.id) == wanted_mode
          Youtube::LivePipeline.schedule_activate!(session) if session.platform == "youtube"
          return
        end

        if running?(session.id) && current_mode(session.id) != wanted_mode
          return unless mode_switch_allowed?(session.id, wanted_mode)

          Rails.logger.info("[OverlayRelay] switch mode #{current_mode(session.id)}→#{wanted_mode} session=#{session.id}")
          stop(session)
        end

        if running?(session.id) && overlay_path_publishing?(session) && current_mode(session.id) == wanted_mode
          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 < 4

          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, mode: wanted_mode)
      rescue Error => e
        Rails.logger.warn("[OverlayRelay] ensure_publishing session=#{session.id}: #{e.message}")
      end

      def desired_mode(session)
        return MODE_COVER if session.paused?
        return MODE_LIVE if Mediamtx::PublisherOnline.video_publishing?(session)
        # RTMP a tratti: resta su live finché c'era segnale telefono (evita flip copertina↔live).
        return MODE_LIVE if session.reconnecting? && live_intake?(session.id)
        return MODE_LIVE if session.status.in?(%w[live connecting reconnecting]) && recent_phone_bytes?(session)

        MODE_COVER
      end

      def terminate_all_session_processes!(session_id)
        session_process_pids(session_id).each do |pid|
          terminate_pid(pid)
        end
        redis_pid = pid_for(session_id).to_i
        terminate_pid(redis_pid) if redis_pid > 1
        clear_pid(session_id)
        clear_mode(session_id)
      end

      def session_process_pids(session_id)
        tag = session_match_tag(session_id)
        pids = []
        each_overlay_process do |pid, cmdline|
          pids << pid if cmdline.include?(tag)
        end
        pids.uniq
      end

      def session_match_tag(session_id)
        "match_#{session_id}"
      end

      def each_overlay_process
        return unless overlay_relay_process_check_local?

        Dir.glob("/proc/[0-9]*").each do |proc_dir|
          pid = proc_dir.split("/").last.to_i
          next if pid <= 1

          cmdline = File.read("#{proc_dir}/cmdline").tr("\0", " ")
          next unless cmdline.include?("ffmpeg") || cmdline.include?("overlay_png_feeder")
          next unless cmdline.include?("match_")

          yield pid, cmdline
        rescue Errno::ENOENT, Errno::EPERM
          nil
        end
      end

      private

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

      def ffmpeg_command(session, pipe, mode)
        mode == MODE_LIVE ? live_command(session, pipe) : cover_command(session, pipe)
      end

      def live_command(session, pipe)
        intake = "#{mediamtx_rtmp_url}/#{session.mediamtx_path_name}"
        [
          "ffmpeg", "-nostdin", "-hide_banner", "-loglevel", "warning",
          "-fflags", "+genpts+discardcorrupt",
          "-analyzeduration", "20000000", "-probesize", "20000000",
          "-rw_timeout", "15000000",
          "-noautorotate",
          "-i", intake,
          "-thread_queue_size", "512",
          "-f", "image2pipe", "-framerate", overlay_input_fps.to_s, "-i", pipe.to_s,
          *common_encode_args(session, live_filter, "0:a?")
        ]
      end

      def cover_command(session, pipe)
        cover = cover_image_path
        raise Error, "copertina fallback non trovata: #{cover}" unless File.exist?(cover)

        [
          "ffmpeg", "-nostdin", "-hide_banner", "-loglevel", "warning",
          "-re", "-loop", "1", "-framerate", output_fps.to_s, "-i", cover,
          "-thread_queue_size", "512",
          "-f", "image2pipe", "-framerate", overlay_input_fps.to_s, "-i", pipe.to_s,
          "-f", "lavfi", "-i", "anullsrc=channel_layout=mono:sample_rate=48000",
          *common_encode_args(session, cover_filter, "2:a")
        ]
      end

      def common_encode_args(session, filter, audio_map)
        fps = output_fps
        [
          "-filter_complex", filter,
          "-map", "[vout]", "-map", audio_map,
          "-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",
          *output_args(session)
        ]
      end

      # Crop center 16:9 a tutto schermo (niente bande nere su TV/web).
      def live_filter
        fps = output_fps
        "[1:v]format=rgba[ol];" \
          "[0:v]scale=1280:720:force_original_aspect_ratio=increase,crop=1280:720,setsar=1,fps=#{fps},format=yuv420p[vid];" \
          "[vid][ol]overlay=0:0:format=auto[vout]"
      end

      def cover_filter
        fps = output_fps
        "[1:v]format=rgba[ol];" \
          "[0:v]scale=1280:720:force_original_aspect_ratio=increase,crop=1280:720,setsar=1,fps=#{fps},format=yuv420p[vid];" \
          "[vid][ol]overlay=0:0:format=auto[vout]"
      end

      def cover_image_path
        ENV.fetch(
          "OVERLAY_COVER_IMAGE",
          Rails.root.join("branding", "CopertinaCanale_6.png").to_s
        )
      end

      # Un solo encode → MediaMTX _air + YouTube RTMPS (evita seconda lettura RTMP senza SPS).
      def output_args(session)
        air = "#{mediamtx_rtmp_url}/#{session.mediamtx_overlay_path_name}"
        if session.platform == "youtube" && session.stream_key.present?
          yt = "rtmps://a.rtmps.youtube.com/live2/#{session.stream_key}"
          tee = "[f=flv:onfail=ignore]#{air}|[f=flv:onfail=ignore:use_fifo=1]#{yt}"
          ["-f", "tee", tee]
        else
          ["-f", "flv", air]
        end
      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), 3000].max
        [(source_kbps * 2.0).to_i, 5500].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["bytesReceived"].to_i > 100_000
      end

      def acquire_start_lock!(session_id)
        redis.set(format(START_LOCK_KEY, session_id), "1", nx: true, ex: 30)
      end

      def release_start_lock!(session_id)
        redis.del(format(START_LOCK_KEY, session_id))
      end

      def publisher_has_video?(session)
        info = Mediamtx::PublisherOnline.path_info(session)
        return false unless Mediamtx::PublisherOnline.active_path?(info)

        (info["tracks2"] || []).any? do |track|
          track["codec"] == "H264" && track.dig("codecProps", "width").to_i.positive?
        end
      end

      def live_intake?(session_id)
        redis.get(format(LIVE_INTAKE_KEY, session_id)) == "1"
      end

      def current_mode(session_id)
        redis.get(format(MODE_KEY, session_id))
      end

      def normalize_mode(mode)
        mode == MODE_LIVE ? MODE_LIVE : MODE_COVER
      end

      def mark_live_intake!(session_id)
        redis.set(format(LIVE_INTAKE_KEY, session_id), "1", ex: 48.hours.to_i)
      end

      def clear_live_intake!(session_id)
        redis.del(format(LIVE_INTAKE_KEY, session_id))
      end

      def store_mode(session_id, mode)
        redis.set(format(MODE_KEY, session_id), normalize_mode(mode), ex: 48.hours.to_i)
      end

      def clear_mode(session_id)
        redis.del(format(MODE_KEY, session_id))
      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 recent_phone_bytes?(session)
        info = Mediamtx::PublisherOnline.path_info(session)
        info && info["bytesReceived"].to_i > 1_000_000
      end

      def publishing_to_youtube?(session)
        return false unless session.platform == "youtube" && session.stream_key.present?

        session_process_pids(session.id).any? do |pid|
          cmdline = File.read("/proc/#{pid}/cmdline").tr("\0", " ")
          cmdline.include?("rtmps://") || cmdline.include?("youtube.com/live2")
        rescue Errno::ENOENT, Errno::EPERM
          false
        end
      end

      def restart_if_missing_youtube_tee!(session)
        return unless session.platform == "youtube" && session.stream_key.present?
        return unless running?(session.id)
        return if publishing_to_youtube?(session)

        Rails.logger.warn("[OverlayRelay] restart overlay: manca tee YouTube session=#{session.id}")
        stop(session)
      end

      def mode_switch_allowed?(session_id, wanted_mode)
        key = format("overlay_relay:mode_want:%s", session_id)
        prev = redis.get(key)
        redis.set(key, wanted_mode, ex: 120)
        return true if prev.blank? || prev == wanted_mode

        ticks = redis.incr(format("overlay_relay:mode_ticks:%s", session_id)).to_i
        redis.expire(format("overlay_relay:mode_ticks:%s", session_id), 60)
        if ticks >= 3
          redis.del(format("overlay_relay:mode_ticks:%s", session_id))
          return true
        end
        false
      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