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"])
      api_base = simulated ? home.api_base_url : "http://#{private_ip}:9997"
      internal_rtmp = simulated ? home.internal_rtmp_url : "rtmp://#{private_ip}:1935"
      internal_hls = simulated ? home.internal_hls_url : "http://#{private_ip}:8888"

      StreamNode.create!(
        slug: slug,
        hostname: hostname,
        role: role,
        status: "ready",
        provider: provider_name_for(cloud, role: role),
        provider_instance_id: instance.id,
        rtmp_base_url: use_node_hostname ? "rtmp://#{hostname}:1935" : home.rtmp_base_url,
        hls_base_url: use_node_hostname ? "https://#{hostname}/hls" : home.hls_base_url,
        api_base_url: api_base,
        internal_rtmp_url: internal_rtmp,
        internal_hls_url: internal_hls,
        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
        }
      )
    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