MatchLiveTv/backend/spec/services/streams/youtube_relay_spec.rb

# frozen_string_literal: true

require "rails_helper"

RSpec.describe Streams::YoutubeRelay do
  describe ".youtube_ffmpeg_args (private)" do
    def args(mode, url)
      described_class.send(:youtube_ffmpeg_args, [mode, url], "rtmps://a.rtmps.youtube.com/live2/key")
    end

    it "remuxes RTMP with video and audio copy (no AAC re-encode)" do
      cmd = args(:rtmp, "rtmp://mediamtx:1935/live/match_x")
      expect(cmd).to include("-c:v", "copy", "-c:a", "copy", "-f", "flv")
      expect(cmd).not_to include("aac")
      expect(cmd.join(" ")).not_to include("-b:a")
    end

    it "remuxes HLS with AAC bitstream filter for FLV" do
      cmd = args(:hls, "http://mediamtx:8888/live/match_x/index.m3u8")
      expect(cmd).to include("-c:v", "copy", "-c:a", "copy", "-bsf:a", "aac_adtstoasc", "-f", "flv")
      expect(cmd).to include("-reconnect", "1")
      expect(cmd).not_to include("-b:a")
    end
  end

  describe "node queue affinity" do
    def with_env(vars)
      previous = vars.keys.index_with { |k| ENV[k] }
      vars.each { |k, v| ENV[k] = v }
      yield
    ensure
      previous.each { |k, v| v.nil? ? ENV.delete(k) : ENV[k] = v }
    end

    let(:session_id) { SecureRandom.uuid }
    let(:lab_node) { instance_double(StreamNode, slug: "ingest-lab-01", role: "lab", relay_agent_url: nil) }
    let(:cloud_node) do
      instance_double(
        StreamNode,
        slug: "ingest-01",
        role: "cloud",
        relay_agent_url: "http://203.0.113.10:9100"
      )
    end

    def session_double(stream_node: nil)
      instance_double(
        StreamSession,
        id: session_id,
        platform: "youtube",
        terminal?: false,
        stream_key: "yt-key",
        status: "live",
        stream_node: stream_node,
        mediamtx_path_name: "live/match_#{session_id}",
        mediamtx_internal_rtmp_url: "rtmp://10.0.0.9:1935",
        mediamtx_internal_hls_url: "http://10.0.0.9:8888"
      )
    end

    it "routes home sessions to youtube_relay_home" do
      expect(described_class.queue_for(session_double)).to eq("youtube_relay_home")
    end

    it "routes overflow lab sessions to the node slug queue" do
      expect(described_class.queue_for(session_double(stream_node: lab_node))).to eq("youtube_relay_ingest-lab-01")
    end

    it "routes cloud overflow to youtube_relay_cloud for the home dispatcher" do
      expect(described_class.queue_for(session_double(stream_node: cloud_node))).to eq("youtube_relay_cloud")
    end

    it "requeues ensure to the assigned node when the local worker does not match" do
      with_env("STREAM_NODE_SLUG" => "home", "YOUTUBE_RELAY_WORKER" => "1") do
        allow(described_class).to receive(:worker?).and_return(true)
        job_proxy = double("ConfiguredJob", perform_later: true)
        expect(YoutubeRelayEnsureJob).to receive(:set).with(hash_including(queue: "youtube_relay_ingest-lab-01", wait: 2.seconds)).and_return(job_proxy)

        expect(described_class.ensure_on_worker!(session_double(stream_node: lab_node))).to eq(:wrong_host)
      end
    end

    it "lets the home worker process cloud overflow instead of requeueing to a per-node queue" do
      with_env("STREAM_NODE_SLUG" => "home", "YOUTUBE_RELAY_WORKER" => "1") do
        expect(described_class.can_process?(session_double(stream_node: cloud_node))).to be true
        expect(described_class.cloud_home_dispatch?(session_double(stream_node: cloud_node))).to be true
      end
    end
  end

  describe "multi-host sticky stop/capacity" do
    let(:redis) { Redis.new(url: ENV.fetch("REDIS_URL", "redis://redis:6379/0")) }
    let(:session_id) { SecureRandom.uuid }

    before do
      allow(described_class).to receive(:worker?).and_return(true)
      allow(described_class).to receive(:worker_id).and_return("worker-a")
      allow(described_class).to receive(:max_concurrent).and_return(1)
      allow(described_class).to receive(:local_node_slug).and_return("home")
      redis.flushdb
    end

    def session_double
      instance_double(
        StreamSession,
        id: session_id,
        platform: "youtube",
        terminal?: false,
        stream_key: "yt-key",
        status: "live",
        stream_node: nil
      )
    end

    it "does not stop ffmpeg owned by another worker" do
      redis.set(format(Streams::YoutubeRelay::OWNER_KEY, session_id), "worker-b", ex: 3600)
      redis.set(format(Streams::YoutubeRelay::REDIS_KEY, session_id), "12345", ex: 3600)

      expect(described_class.stop_on_worker!(session_double)).to eq(:wrong_host)
      expect(redis.get(format(Streams::YoutubeRelay::OWNER_KEY, session_id))).to eq("worker-b")
    end

    it "requeues ensure on the node queue when at capacity" do
      other_id = SecureRandom.uuid
      redis.sadd(format(Streams::YoutubeRelay::OWNED_SET, "worker-a"), other_id)
      allow(described_class).to receive(:intake_available?).and_return(true)
      expect(YoutubeRelayEnsureJob).to receive(:set).with(hash_including(wait: 5.seconds, queue: "youtube_relay_home")).and_return(
        double(perform_later: true)
      )

      expect(described_class.ensure_on_worker!(session_double)).to eq(:at_capacity)
    end
  end

  describe ".mediamtx_intake_source (private)" do
    def with_env(vars)
      previous = vars.keys.index_with { |k| ENV[k] }
      vars.each { |k, v| ENV[k] = v }
      yield
    ensure
      previous.each { |k, v| v.nil? ? ENV.delete(k) : ENV[k] = v }
    end

    let(:session) do
      instance_double(
        StreamSession,
        stream_node: instance_double(StreamNode, slug: "ingest-lab-01", role: "lab", relay_agent_url: nil),
        mediamtx_path_name: "live/match_abc",
        mediamtx_internal_rtmp_url: "rtmp://10.0.0.9:1935",
        mediamtx_internal_hls_url: "http://10.0.0.9:8888"
      )
    end

    it "uses loopback RTMP when the worker is on the assigned node" do
      with_env(
        "STREAM_NODE_SLUG" => "ingest-lab-01",
        "STREAM_NODE_LOCAL_RTMP_URL" => "rtmp://127.0.0.1:1935"
      ) do
        allow(Mediamtx::PublisherOnline).to receive(:active?).with(session).and_return(true)
        expect(described_class.send(:mediamtx_intake_source, session)).to eq(
          [:rtmp, "rtmp://127.0.0.1:1935/live/match_abc"]
        )
      end
    end

    it "uses loopback when the node relay agent is enabled" do
      with_env(
        "STREAM_NODE_SLUG" => "ingest-lab-01",
        "STREAM_NODE_RELAY_AGENT_URL" => "http://203.0.113.10:9100",
        "STREAM_NODE_LOCAL_RTMP_URL" => "rtmp://mediamtx_lab:1935"
      ) do
        allow(Mediamtx::PublisherOnline).to receive(:active?).with(session).and_return(true)
        expect(described_class.send(:mediamtx_intake_source, session)).to eq(
          [:rtmp, "rtmp://127.0.0.1:1935/live/match_abc"]
        )
      end
    end
  end

  describe "relay agent spawn" do
    def with_env(vars)
      previous = vars.keys.index_with { |k| ENV[k] }
      vars.each { |k, v| ENV[k] = v }
      yield
    ensure
      previous.each { |k, v| v.nil? ? ENV.delete(k) : ENV[k] = v }
    end

    it "asks the node agent to start ffmpeg and stores the pid" do
      session = instance_double(
        StreamSession,
        id: "sess-1",
        mediamtx_path_name: "live/match_abc",
        stream_key: "yt-key"
      )
      allow(described_class).to receive(:log_file).and_return("/tmp/youtube_relay_test.log")
      allow(File).to receive(:write)
      expect(described_class).to receive(:agent_request).with(
        Net::HTTP::Post,
        "/relays",
        hash_including("session_id" => "sess-1", "path" => "live/match_abc"),
        session: session
      ).and_return("pid" => 4242)

      pid = with_env("STREAM_NODE_RELAY_AGENT_URL" => "http://203.0.113.10:9100") do
        described_class.send(:start_via_agent!, session)
      end
      expect(pid).to eq(4242)
    end
  end
end

RSpec.describe StreamNode do
  describe "#relay_agent_url" do
    it "uses metadata public_ip on cloud nodes" do
      node = described_class.new(
        slug: "ingest-01",
        hostname: "ingest-01.mltv-stream.net",
        role: "cloud",
        status: "ready",
        provider: "hetzner",
        rtmp_base_url: "rtmp://x:1935",
        hls_base_url: "https://x/hls",
        api_base_url: "http://10.0.0.9:9997",
        metadata: { "public_ip" => "203.0.113.9" }
      )
      expect(node.relay_agent_url).to eq("http://203.0.113.9:9100")
    end

    it "is nil for lab nodes" do
      node = described_class.new(role: "lab", api_base_url: "http://10.0.0.9:9997", metadata: { "public_ip" => "1.2.3.4" })
      expect(node.relay_agent_url).to be_nil
    end
  end
end