MatchLiveTv/backend/lib/tasks/streams_nodes.rake

# frozen_string_literal: true

namespace :streams do
  desc "Chiude CPX cloud idle in quiet hours (02:00–07:00 Europe/Rome); alert se sessioni attive"
  task night_cloud_sweep: :environment do
    result = Streams::NightCloudSweeper.sweep!
    puts "skipped=#{result.skipped} error=#{result.error} actions=#{result.actions.inspect}"
  end

  namespace :nodes do
    desc "Assicura il nodo home dagli ENV MediaMTX"
    task ensure_home: :environment do
      node = Streams::NodeRegistry.ensure_home_from_env!
      puts "home ready slug=#{node.slug} rtmp=#{node.rtmp_base_url} max=#{node.max_publishers}"
    end

    desc "Provisiona un nodo lab (STREAM_CLOUD_PROVIDER=local_lab|proxmox_lab)"
    task provision_lab: :environment do
      node = Streams::NodeProvisioner.new.provision_lab!
      puts "lab node ready slug=#{node.slug} host=#{node.hostname} id=#{node.provider_instance_id}"
      if ENV.fetch("STREAM_DNS_PROVIDER", "lab") == "lab"
        puts "DNS lab snippet:"
        puts Streams::DnsProviders::Lab.new.hosts_file_snippet
      end
    end

    desc "Provisiona un nodo Hetzner Cloud + DNS mltv-stream.net (richiede HCLOUD_TOKEN)"
    task provision_cloud: :environment do
      node = Streams::NodeProvisioner.new.provision_cloud!
      puts "cloud node ready slug=#{node.slug} host=#{node.hostname} id=#{node.provider_instance_id} ip=#{node.metadata['public_ip']}"
    end

    desc "Dump record DNS lab (Redis)"
    task dns_lab_dump: :environment do
      puts Streams::DnsProviders::Lab.new.hosts_file_snippet
    end
  end

  namespace :relay do
    desc "Mostra coda e slug nodo per i worker ffmpeg YouTube"
    task queues: :environment do
      puts "local_slug=#{Streams::YoutubeRelay.local_node_slug} worker=#{Streams::YoutubeRelay.worker?}"
      StreamNode.order(:slug).find_each do |node|
        puts "node=#{node.slug} queue=youtube_relay_#{node.slug} status=#{node.status} relays=#{node.max_relays}"
      end
    end

    desc "Simula un telefono RTMP e verifica che ffmpeg parta sul worker del nodo lab"
    task simulate: :environment do
      abort "Esegui questo task nel container Rails (docker compose exec rails ...)" unless File.exist?("/.dockerenv") || ENV["FORCE_RELAY_SIMULATE"] == "1"

      home = Streams::NodeRegistry.ensure_home_from_env!
      lab = StreamNode.find_by(slug: ENV.fetch("STREAM_NODE_SLUG", "ingest-lab-01"))
      lab ||= Streams::NodeProvisioner.new.provision_lab!
      user = User.find_by!(email: ENV.fetch("E2E_USER_EMAIL", "coach@matchlivetv.test"))
      team = user.teams.first || user.clubs.first&.teams&.first || Team.first
      abort "Nessun team per #{user.email}" unless team
      match = team.matches.first || team.matches.create!(opponent_name: "Relay sim")

      previous_max = home.max_publishers
      filler = nil
      session = nil
      ffmpeg_pid = nil
      begin
        home.update!(max_publishers: 1, status: "ready")
        filler = occupy_home!(home, user, match)
        session = build_lab_youtube_session!(lab, user, match)
        Mediamtx::Client.for_session(session).create_path(session)
        session.begin_connect! if session.may_begin_connect?
        session.update!(status: "connecting") unless session.status.in?(%w[connecting live reconnecting])

        puts "filler_home=#{filler.id} lab_session=#{session.id} node=#{session.stream_node.slug} queue=#{Streams::YoutubeRelay.queue_for(session)}"
        ffmpeg_pid = spawn_test_rtmp!(session)
        wait_for!("publisher MediaMTX", 30) { Mediamtx::PublisherOnline.active?(session.reload) }

        Streams::YoutubeRelay.enqueue_ensure!(session)
        wait_for!("owner Redis sul nodo lab", 25) do
          owner_for(session.id) == lab.slug
        end

        owner = owner_for(session.id)
        home_owns = relay_redis.sismember(format(Streams::YoutubeRelay::OWNED_SET, "home"), session.id)
        puts "owner=#{owner} home_owns=#{home_owns} queue=#{Streams::YoutubeRelay.queue_for(session)}"
        abort "FAIL: owner atteso #{lab.slug}, trovato #{owner.inspect}" unless owner == lab.slug
        abort "FAIL: il worker home ha reclamato la sessione lab" if home_owns
        puts "OK — ffmpeg assegnato al worker #{lab.slug}, non a home"
      ensure
        begin
          Process.kill("TERM", ffmpeg_pid) if ffmpeg_pid
        rescue Errno::ESRCH
          nil
        end
        session&.update!(status: "ended", ended_at: Time.current) if session&.persisted?
        filler&.update!(status: "ended", ended_at: Time.current) if filler&.persisted?
        home.update!(max_publishers: previous_max)
      end
    end
  end
end

def occupy_home!(home, user, match)
  StreamSession.create!(
    match: match,
    user: user,
    platform: "matchlivetv",
    status: "live",
    privacy_status: "unlisted",
    stream_node: home,
    publish_token: SecureRandom.hex(8),
    started_at: Time.current
  )
end

def build_lab_youtube_session!(lab, user, match)
  StreamSession.create!(
    match: match,
    user: user,
    platform: "youtube",
    status: "idle",
    privacy_status: "unlisted",
    stream_node: lab,
    publish_token: SecureRandom.hex(8),
    stream_key: "e2e-dummy-#{SecureRandom.hex(6)}"
  )
end

def spawn_test_rtmp!(session)
  intake = "#{session.mediamtx_internal_rtmp_url.chomp('/')}/#{session.mediamtx_path_name}"
  cmd = [
    "ffmpeg", "-hide_banner", "-loglevel", "error",
    "-re", "-f", "lavfi", "-i", "testsrc=size=1280x720:rate=30",
    "-f", "lavfi", "-i", "sine=frequency=440:sample_rate=48000",
    "-t", "40",
    "-c:v", "libx264", "-preset", "veryfast", "-pix_fmt", "yuv420p", "-b:v", "1500k",
    "-c:a", "aac", "-ac", "1", "-ar", "48000", "-b:a", "96k",
    "-f", "flv", intake
  ]
  pid = Process.spawn(*cmd, %i[out err] => "/tmp/relay_sim_ffmpeg.log")
  Process.detach(pid)
  puts "telefono simulato pid=#{pid} rtmp=#{intake}"
  pid
end

def wait_for!(label, timeout_sec)
  deadline = Time.now + timeout_sec
  loop do
    return true if yield
    abort "FAIL timeout: #{label}" if Time.now >= deadline

    sleep 1
  end
end

def owner_for(session_id)
  relay_redis.get(format(Streams::YoutubeRelay::OWNER_KEY, session_id))
end

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