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

# frozen_string_literal: true

require "rails_helper"

RSpec.describe Streams::Autoscaler 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(:redis) { Redis.new(url: ENV.fetch("REDIS_URL", "redis://redis:6379/0")) }

  before do
    redis.del(Streams::Autoscaler::LOCK_KEY)
    redis.del(Streams::Autoscaler::KILL_SWITCH_KEY)
    redis.del(Streams::DnsProviders::Lab::REDIS_KEY)
    StreamSession.where.not(stream_node_id: nil).update_all(stream_node_id: nil, status: "ended", ended_at: Time.current)
    StreamNode.where.not(slug: Streams::NodeRegistry::HOME_SLUG).delete_all
  end

  it "is a no-op when disabled" do
    with_env("STREAM_AUTOSCALE_ENABLED" => "0") do
      result = described_class.reconcile!
      expect(result.skipped).to eq(true)
      expect(result.actions).to eq([])
    end
  end

  it "scales out and creates a warm spare when free slots are low" do
    with_env(
      "STREAM_AUTOSCALE_ENABLED" => "1",
      "STREAM_AUTOSCALE_SOFT_FREE_SLOTS" => "2",
      "STREAM_AUTOSCALE_WARM_SPARE" => "1",
      "STREAM_AUTOSCALE_KIND" => "lab",
      "STREAM_AUTOSCALE_MAX_NODES" => "5",
      "STREAM_CLOUD_PROVIDER" => "local_lab",
      "STREAM_DNS_PROVIDER" => "lab",
      "STREAM_NODE_HOME_MAX_PUBLISHERS" => "2",
      "MEDIAMTX_API_URL" => "http://mtx-home:9997",
      "MEDIAMTX_RTMP_URL" => "rtmp://home.example:1935",
      "HLS_PUBLIC_URL" => "https://home.example/hls"
    ) do
      home = Streams::NodeRegistry.ensure_home_from_env!
      user = User.create!(email: "as@example.com", name: "A", password: "Password123", role: "coach")
      club = Club.create!(name: "AS", sport: "volleyball")
      team = club.teams.create!(name: "T", sport: "volleyball", slug: "as-t")
      match = team.matches.create!(opponent_name: "X", scheduled_at: 1.hour.from_now)
      # Fill home (2/2) → free_slots=0
      2.times do
        StreamSession.create!(match: match, user: user, platform: "matchlivetv", status: "live", stream_node: home)
      end

      provisioner = instance_double(Streams::NodeProvisioner)
      created = []
      allow(provisioner).to receive(:provision_lab!) do
        node = StreamNode.create!(
          slug: "ingest-lab-#{created.size + 1}",
          hostname: "h#{created.size}.lab",
          role: "lab",
          status: "ready",
          provider: "local",
          rtmp_base_url: "rtmp://h:1935",
          hls_base_url: "https://h/hls",
          api_base_url: "http://h:9997",
          max_publishers: 2,
          max_relays: 2
        )
        created << node
        node
      end

      result = described_class.reconcile!(provisioner: provisioner)
      expect(result.actions).to include(:scale_out)
      expect(created.size).to eq(1)
      expect(result.metrics[:free_slots]).to be >= 2

      # Riempi il nodo overflow → spare_ready=0 ma carico presente → warm spare
      StreamSession.create!(
        match: match, user: user, platform: "matchlivetv", status: "live", stream_node: created.first
      )
      StreamSession.create!(
        match: match, user: user, platform: "matchlivetv", status: "live", stream_node: created.first
      )

      result2 = described_class.reconcile!(provisioner: provisioner)
      expect(result2.actions).to include(:warm_spare).or include(:scale_out)
      expect(created.size).to be >= 2
    end
  end

  it "does not scale in a node provisioned in the same reconcile round" do
    with_env(
      "STREAM_AUTOSCALE_ENABLED" => "1",
      "STREAM_AUTOSCALE_SOFT_FREE_SLOTS" => "2",
      "STREAM_AUTOSCALE_WARM_SPARE" => "0",
      "STREAM_AUTOSCALE_IDLE_MINUTES" => "0",
      "STREAM_AUTOSCALE_KIND" => "lab",
      "STREAM_AUTOSCALE_MAX_NODES" => "5",
      "STREAM_CLOUD_PROVIDER" => "local_lab",
      "STREAM_DNS_PROVIDER" => "lab",
      "STREAM_NODE_HOME_MAX_PUBLISHERS" => "2",
      "MEDIAMTX_API_URL" => "http://mtx-home:9997",
      "MEDIAMTX_RTMP_URL" => "rtmp://home.example:1935",
      "HLS_PUBLIC_URL" => "https://home.example/hls"
    ) do
      home = Streams::NodeRegistry.ensure_home_from_env!
      user = User.create!(email: "same@example.com", name: "S", password: "Password123", role: "coach")
      club = Club.create!(name: "Same", sport: "volleyball")
      team = club.teams.create!(name: "T", sport: "volleyball", slug: "same-t")
      match = team.matches.create!(opponent_name: "X", scheduled_at: 1.hour.from_now)
      2.times do
        StreamSession.create!(match: match, user: user, platform: "matchlivetv", status: "live", stream_node: home)
      end

      provisioner = instance_double(Streams::NodeProvisioner)
      created = nil
      allow(provisioner).to receive(:provision_lab!) do
        created = StreamNode.create!(
          slug: "ingest-lab-new",
          hostname: "new.lab",
          role: "lab",
          status: "ready",
          provider: "local",
          rtmp_base_url: "rtmp://h:1935",
          hls_base_url: "https://h/hls",
          api_base_url: "http://h:9997",
          max_publishers: 4,
          max_relays: 2
        )
      end
      expect(provisioner).not_to receive(:decommission!)

      result = described_class.reconcile!(provisioner: provisioner)
      expect(result.actions).to include(:scale_out)
      expect(result.actions.map(&:to_s)).not_to include("scale_in_ingest-lab-new")
      expect(StreamNode.find_by(slug: "ingest-lab-new")).to be_present
    end
  end

  it "scales in an idle overflow node when warm spare is not required" do
    with_env(
      "STREAM_AUTOSCALE_ENABLED" => "1",
      "STREAM_AUTOSCALE_SOFT_FREE_SLOTS" => "0",
      "STREAM_AUTOSCALE_WARM_SPARE" => "0",
      "STREAM_AUTOSCALE_IDLE_MINUTES" => "0",
      "STREAM_NODE_HOME_MAX_PUBLISHERS" => "6",
      "MEDIAMTX_API_URL" => "http://mtx-home:9997",
      "MEDIAMTX_RTMP_URL" => "rtmp://home.example:1935",
      "HLS_PUBLIC_URL" => "https://home.example/hls"
    ) do
      Streams::NodeRegistry.ensure_home_from_env!
      idle = StreamNode.create!(
        slug: "ingest-lab-99",
        hostname: "idle.lab",
        role: "lab",
        status: "ready",
        provider: "local",
        provider_instance_id: "sim-x",
        rtmp_base_url: "rtmp://h:1935",
        hls_base_url: "https://h/hls",
        api_base_url: "http://h:9997",
        max_publishers: 2,
        max_relays: 2,
        created_at: 2.hours.ago
      )

      provisioner = instance_double(Streams::NodeProvisioner)
      allow(provisioner).to receive(:drain!)
      expect(provisioner).to receive(:decommission!).with(idle)

      result = described_class.reconcile!(provisioner: provisioner)
      expect(result.actions.map(&:to_s)).to include("scale_in_ingest-lab-99")
    end
  end

  it "skips when Redis kill-switch is engaged even if ENV is enabled" do
    redis.del(described_class::KILL_SWITCH_KEY)
    with_env("STREAM_AUTOSCALE_ENABLED" => "1") do
      described_class.engage_kill_switch!
      expect(described_class.enabled?).to eq(false)
      result = described_class.reconcile!
      expect(result.skipped).to eq(true)
    ensure
      described_class.clear_kill_switch!
    end
  end

  it "blocks scale-out when next node would exceed monthly budget" do
    with_env(
      "STREAM_AUTOSCALE_ENABLED" => "1",
      "STREAM_AUTOSCALE_SOFT_FREE_SLOTS" => "2",
      "STREAM_AUTOSCALE_WARM_SPARE" => "0",
      "STREAM_AUTOSCALE_KIND" => "lab",
      "STREAM_AUTOSCALE_MAX_NODES" => "5",
      "STREAM_AUTOSCALE_NODE_EUR_PER_HOUR" => "1",
      "STREAM_AUTOSCALE_MONTHLY_BUDGET_EUR" => "10",
      "STREAM_CLOUD_PROVIDER" => "local_lab",
      "STREAM_DNS_PROVIDER" => "lab",
      "STREAM_NODE_HOME_MAX_PUBLISHERS" => "1",
      "MEDIAMTX_API_URL" => "http://mtx-home:9997",
      "MEDIAMTX_RTMP_URL" => "rtmp://home.example:1935",
      "HLS_PUBLIC_URL" => "https://home.example/hls"
    ) do
      StreamSession.where.not(stream_node_id: nil).update_all(stream_node_id: nil, status: "ended", ended_at: Time.current)
      StreamNode.where.not(slug: Streams::NodeRegistry::HOME_SLUG).delete_all
      home = Streams::NodeRegistry.ensure_home_from_env!
      home.update!(max_publishers: 1)
      user = User.create!(email: "budget@example.com", name: "B", password: "Password123", role: "coach")
      club = Club.create!(name: "Budget", sport: "volleyball")
      team = club.teams.create!(name: "T", sport: "volleyball", slug: "budget-t")
      match = team.matches.create!(opponent_name: "X", scheduled_at: 1.hour.from_now)
      StreamSession.create!(match: match, user: user, platform: "matchlivetv", status: "live", stream_node: home)

      provisioner = instance_double(Streams::NodeProvisioner)
      expect(provisioner).not_to receive(:provision_lab!)

      result = described_class.reconcile!(provisioner: provisioner)
      expect(result.actions).to include(:blocked_capacity)
      # 0 overflow nodes → stima 0€ ≤ budget; il blocco è sul nodo successivo (+1)
      expect(described_class.within_budget?(0)).to eq(true)
      expect(described_class.within_budget?(1)).to eq(false)
    end
  end

  it "promotes provisioning nodes when MediaMTX becomes reachable" do
    with_env(
      "STREAM_AUTOSCALE_ENABLED" => "1",
      "STREAM_AUTOSCALE_WARM_SPARE" => "0",
      "STREAM_AUTOSCALE_SOFT_FREE_SLOTS" => "0",
      "STREAM_AUTOSCALE_KIND" => "lab",
      "MEDIAMTX_API_URL" => "http://mtx-home:9997",
      "MEDIAMTX_RTMP_URL" => "rtmp://home.example:1935",
      "HLS_PUBLIC_URL" => "https://home.example/hls"
    ) do
      Streams::NodeRegistry.ensure_home_from_env!
      node = StreamNode.create!(
        slug: "ingest-lab-prov",
        hostname: "prov.lab",
        role: "lab",
        status: "provisioning",
        provider: "local",
        rtmp_base_url: "rtmp://h:1935",
        hls_base_url: "https://h/hls",
        api_base_url: "http://h:9997",
        max_publishers: 2,
        max_relays: 2
      )
      allow(Streams::NodeHealth).to receive(:promote_if_healthy!) do |n|
        n.update!(status: "ready", last_health_at: Time.current)
        true
      end

      provisioner = instance_double(Streams::NodeProvisioner)
      expect(provisioner).not_to receive(:provision_lab!)

      result = described_class.reconcile!(provisioner: provisioner)
      expect(result.actions).to include(:"ready_ingest-lab-prov")
      expect(node.reload.status).to eq("ready")
    end
  end
end