MatchLiveTv/backend/app/jobs/overlay_relay_ensure_job.rb

# Avvia overlay ffmpeg solo in Sidekiq (un solo job per sessione alla volta).
class OverlayRelayEnsureJob < ApplicationJob
  queue_as :default

  LOCK_KEY = "overlay_ensure:lock:%s"
  LOCK_TTL = 55
  RETRY_WAIT = 5.seconds

  class << self
    def enqueue_unique(session_id, attempt: 1, force_restart: false, wait: 0.seconds)
      key = format(LOCK_KEY, session_id)
      r = Redis.new(url: ENV.fetch("REDIS_URL", "redis://localhost:6379/0"))
      return unless r.set(key, "1", nx: true, ex: LOCK_TTL)

      set(wait: wait).perform_later(session_id, attempt, force_restart: force_restart)
    end
  end

  def perform(session_id, attempt = 1, force_restart: false)
    session = StreamSession.find_by(id: session_id)
    unless session && !session.terminal?
      clear_lock(session_id)
      return
    end
    return unless session.status.in?(%w[live connecting reconnecting paused])

    begin
      run_ensure(session, session_id, attempt, force_restart)
    ensure
      clear_lock(session_id)
    end
  end

  private

  def run_ensure(session, session_id, attempt, force_restart)
    if Mediamtx::PublisherOnline.video_publishing?(session) && !session.paused?
      session.go_live! if session.may_go_live?
      session.reconnect! if session.reconnecting? && session.may_reconnect?
      Mediamtx::Client.new.set_always_available(session, enabled: false)
    end

    if force_restart
      Streams::OverlayRelay.stop(session)
      Streams::OverlayRelay.terminate_all_session_processes!(session.id)
    end

    Streams::OverlayRelay.ensure_publishing!(session)

    if Streams::OverlayRelay.healthy?(session)
      Youtube::LivePipeline.schedule_activate!(session, force: true, wait: 5.seconds)
      return
    end

    return if Streams::OverlayRelay.running?(session.id)

    Rails.logger.info("[OverlayRelayEnsureJob] overlay not running, retry session=#{session_id} attempt=#{attempt}")
    reschedule(session_id, attempt, false) if attempt < 30
  rescue Mediamtx::Client::Error => e
    Rails.logger.warn("[OverlayRelayEnsureJob] session=#{session_id}: #{e.message}")
  end

  def reschedule(session_id, attempt, force_restart)
    clear_lock(session_id)
    self.class.enqueue_unique(
      session_id,
      attempt: attempt + 1,
      force_restart: force_restart,
      wait: RETRY_WAIT
    )
  end

  def clear_lock(session_id)
    redis.del(format(LOCK_KEY, session_id))
  end

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