MatchLiveTv/backend/app/services/streams/node_provisioner.rb

# frozen_string_literal: true

module Streams
  # Provisiona / decommissiona nodi stream (lab o cloud) e aggiorna DNS + registry.
  class NodeProvisioner
    class Error < StandardError; end
    class BusyError < Error; end

    LAB_DNS_SUFFIX = -> { ENV.fetch("STREAM_LAB_DNS_SUFFIX", "lab.mltv-stream.net") }
    CLOUD_DNS_SUFFIX = -> { ENV.fetch("STREAM_CLOUD_DNS_SUFFIX", ENV.fetch("STREAM_DNS_ZONE", "mltv-stream.net")) }

    def initialize(cloud: nil, dns: nil)
      @cloud = cloud
      @dns = dns
    end

    def provision_lab!(prefix: "ingest-lab")
      provision!(
        prefix: prefix,
        role: "lab",
        dns_suffix: LAB_DNS_SUFFIX.call,
        cloud: cloud_provider(ENV.fetch("STREAM_CLOUD_PROVIDER", "local_lab")),
        dns: dns_provider(ENV.fetch("STREAM_DNS_PROVIDER", "lab")),
        max: ENV.fetch("STREAM_LAB_MAX_PUBLISHERS", "2").to_i,
        use_node_hostname: ENV["STREAM_LAB_USE_NODE_HOSTNAME"] == "1"
      )
    end

    def provision_cloud!(prefix: "ingest")
      provision!(
        prefix: prefix,
        role: "cloud",
        dns_suffix: CLOUD_DNS_SUFFIX.call,
        cloud: cloud_provider("hetzner"),
        dns: dns_provider("hetzner"),
        max: ENV.fetch("STREAM_CLOUD_MAX_PUBLISHERS", "4").to_i,
        use_node_hostname: true
      )
    end

    def decommission!(node)
      raise Error, "Non si può decommissionare il nodo home" if node.slug == Streams::NodeRegistry::HOME_SLUG
      if node.occupying_sessions.exists?
        raise BusyError, "Nodo #{node.slug} ha ancora sessioni attive"
      end

      node.update!(status: "draining")
      cloud = cloud_for_node(node)
      dns = dns_for_node(node)
      cloud.destroy_node(node.provider_instance_id) if node.provider_instance_id.present?
      dns.delete_a(node.hostname) if node.hostname.present?
      node.destroy!
      true
    end

    def drain!(node)
      raise Error, "Non si può mettere in drain il nodo home" if node.slug == Streams::NodeRegistry::HOME_SLUG

      node.update!(status: "draining")
      node
    end

    private

    def provision!(prefix:, role:, dns_suffix:, cloud:, dns:, max:, use_node_hostname:)
      Streams::NodeRegistry.ensure_home_from_env!
      home = StreamNode.find_by!(slug: Streams::NodeRegistry::HOME_SLUG)
      slug = next_slug(prefix)
      hostname = "#{slug}.#{dns_suffix}"

      instance = cloud.create_node(
        name: "mltv-stream-#{slug}",
        labels: { role: "stream-node", env: role == "cloud" ? "prod" : "lab" }
      )
      ip = instance.public_ip.presence || "127.0.0.1"
      private_ip = instance.private_ip.presence || ip
      dns.upsert_a(hostname, ip)

      simulated = instance.raw.is_a?(Hash) && (instance.raw[:simulated] || instance.raw["simulated"])
      urls = urls_for_node(role: role, home: home, hostname: hostname, simulated: simulated,
                           private_ip: private_ip, public_ip: ip, use_node_hostname: use_node_hostname)

      # Simulated/lab che riusa MediaMTX home: subito ready. Cloud reale: provisioning
      # finché :9997 risponde (evita allocate → Faraday Connection refused → 500).
      initial_status = simulated ? "ready" : "provisioning"

      node = StreamNode.create!(
        slug: slug,
        hostname: hostname,
        role: role,
        status: initial_status,
        provider: provider_name_for(cloud, role: role),
        provider_instance_id: instance.id,
        rtmp_base_url: urls.fetch(:rtmp_base_url),
        hls_base_url: urls.fetch(:hls_base_url),
        api_base_url: urls.fetch(:api_base_url),
        internal_rtmp_url: urls.fetch(:internal_rtmp_url),
        internal_hls_url: urls.fetch(:internal_hls_url),
        max_publishers: max,
        max_relays: max,
        last_health_at: Time.current,
        metadata: {
          "public_ip" => ip,
          "private_ip" => private_ip,
          "simulated" => simulated,
          "cloud_raw" => instance.raw
        }
      )

      wait_until_mediamtx_ready!(node) unless simulated
      node.reload
    end

    def wait_until_mediamtx_ready!(node, timeout: nil, interval: nil)
      timeout ||= ENV.fetch("STREAM_NODE_READY_TIMEOUT_SECS", "180").to_i
      interval ||= ENV.fetch("STREAM_NODE_READY_POLL_SECS", "3").to_f
      deadline = Process.clock_gettime(Process::CLOCK_MONOTONIC) + timeout

      loop do
        if Streams::NodeHealth.promote_if_healthy!(node)
          Rails.logger.info("[Streams::NodeProvisioner] #{node.slug} MediaMTX ready")
          return node
        end

        remaining = deadline - Process.clock_gettime(Process::CLOCK_MONOTONIC)
        if remaining <= 0
          Rails.logger.warn(
            "[Streams::NodeProvisioner] #{node.slug} ancora provisioning dopo #{timeout}s " \
            "(api=#{node.api_base_url}); autoscaler continuerà a promuovere"
          )
          return node
        end

        sleep([interval, remaining].min)
        node.reload
      end
    end

    def urls_for_node(role:, home:, hostname:, simulated:, private_ip:, public_ip: nil, use_node_hostname:)
      if role == "lab" && ENV["MEDIAMTX_LAB_API_URL"].present?
        return {
          api_base_url: ENV.fetch("MEDIAMTX_LAB_API_URL"),
          internal_rtmp_url: ENV.fetch("MEDIAMTX_LAB_INTERNAL_RTMP_URL", "rtmp://mediamtx_lab:1935"),
          internal_hls_url: ENV.fetch("MEDIAMTX_LAB_HLS_URL", "http://mediamtx_lab:8888"),
          rtmp_base_url: ENV.fetch("MEDIAMTX_LAB_RTMP_URL", "rtmp://127.0.0.1:11935"),
          # HLS pubblico resta sul proxy del sito: il controller instrada al MediaMTX del nodo.
          hls_base_url: home.hls_base_url
        }
      end

      control_ip = control_ip_for(role: role, private_ip: private_ip, public_ip: public_ip)
      {
        api_base_url: simulated ? home.api_base_url : "http://#{control_ip}:9997",
        internal_rtmp_url: simulated ? home.internal_rtmp_url : "rtmp://#{control_ip}:1935",
        internal_hls_url: simulated ? home.internal_hls_url : "http://#{control_ip}:8888",
        rtmp_base_url: use_node_hostname ? "rtmp://#{hostname}:1935" : home.rtmp_base_url,
        # Player del sito: proxy Rails/edge. HTTPS diretto sul nodo arriva dopo TLS/Caddy.
        hls_base_url: home.hls_base_url
      }
    end

    # Senza WireGuard/network privata Rails deve parlare con l'IPv4 pubblico (API+HLS).
    def control_ip_for(role:, private_ip:, public_ip:)
      public_ip = public_ip.presence
      return private_ip if role != "cloud" || public_ip.blank?
      return public_ip if ENV["STREAM_CLOUD_PUBLIC_CONTROL"] == "1"
      return public_ip if private_ip.blank? || private_ip == public_ip

      private_ip
    end

    def next_slug(prefix)
      used = StreamNode.where("slug LIKE ?", "#{prefix}-%").pluck(:slug)
      n = 1
      loop do
        candidate = format("%s-%02d", prefix, n)
        return candidate unless used.include?(candidate)

        n += 1
      end
    end

    def cloud_provider(name)
      @cloud || Streams::CloudProviders.build(name)
    end

    def dns_provider(name)
      @dns || Streams::DnsProviders.build(name)
    end

    def provider_name_for(cloud, role: nil)
      return "hetzner" if role.to_s == "cloud"

      case cloud
      when Streams::CloudProviders::ProxmoxLab then "proxmox_lab"
      when Streams::CloudProviders::Hetzner then "hetzner"
      else "local"
      end
    end

    def cloud_for_node(node)
      case node.provider
      when "hetzner" then Streams::CloudProviders::Hetzner.new
      when "proxmox_lab" then Streams::CloudProviders::ProxmoxLab.new
      else Streams::CloudProviders::LocalLab.new
      end
    end

    def dns_for_node(node)
      case node.provider
      when "hetzner" then Streams::DnsProviders::Hetzner.new
      else Streams::DnsProviders::Lab.new
      end
    end
  end
end