namespace :youtube do
  desc "Test pipeline YouTube (setup → RTMP test → relay copy → activate). SESSION_ID= opzionale"
  task e2e: :environment do
    abort "YOUTUBE_CLIENT_ID mancante" if ENV["YOUTUBE_CLIENT_ID"].blank?

    session = resolve_e2e_session!
    puts "Sessione: #{session.id} (#{session.match.team.name} vs #{session.match.opponent_name})"
    puts "Stato iniziale: #{session.status} broadcast=#{session.youtube_broadcast_id.present?} key=#{session.stream_key.present?}"

    if Youtube::ApiThrottle.rate_limited?
      puts "ATTENZIONE: rate limit attivo, attendo 120s..."
      sleep 120
      clear_youtube_throttle!
    end

    unless session.youtube_broadcast_id.present? && session.stream_key.present?
      puts "→ Setup broadcast YouTube..."
      setup_with_retries!(session)
      session.reload
      abort "Setup fallito: #{session.youtube_broadcast_id.inspect} / key=#{session.stream_key.present?}" unless session.stream_key.present?
    end

    puts "→ Watch URL: #{session.youtube_watch_url}"

    session.begin_connect! if session.may_begin_connect?
    session.update!(status: "connecting") unless session.status.in?(%w[connecting live reconnecting])

    puts "→ Simulo telefono (RTMP test pattern 60s)..."
    push_test_rtmp!(session, duration: 60)

    wait_for("publisher video su MediaMTX", 45) { Mediamtx::PublisherOnline.video_publishing?(session.reload) }

    Youtube::LivePipeline.schedule!(session, force: true)
    wait_for("relay YouTube attivo", 90) { Streams::YoutubeRelay.running?(session.id) }
    wait_for("ingest YouTube active", 120) { youtube_ingest_active?(session) }

    Youtube::LivePipeline.schedule_activate!(session, force: true, wait: 5.seconds)

    result = nil
    wait_for("broadcast live", 180) do
      result = Youtube::BroadcastService.new(session.match.team).activate_broadcast!(session.youtube_broadcast_id)
      %i[live transitioned].include?(result)
    end

    puts "Risultato activate: #{result}"
    puts "Ingest: #{ingest_status(session)}"
    puts "Relay: #{relay_summary(session)}"
    puts session.terminal? ? "Sessione terminata." : "OK — YouTube verificato."
  end

  desc "Termina sessioni YouTube di test più vecchie di 2h"
  task cleanup_test: :environment do
    old = StreamSession.where(platform: "youtube", status: %w[connecting live reconnecting paused idle])
                       .where("created_at < ?", 2.hours.ago)
    old.find_each do |s|
      Sessions::Stop.new(s).call
      puts "ended #{s.id}"
    rescue StandardError => e
      puts "skip #{s.id}: #{e.message}"
    end
  end

  def clear_youtube_throttle!
    Redis.new(url: ENV.fetch("REDIS_URL", "redis://localhost:6379/0")).del(Youtube::ApiThrottle::KEY)
  end

  def setup_with_retries!(session, max_attempts: 10)
    max_attempts.times do |i|
      clear_youtube_throttle!

      Youtube::SetupBroadcast.call(session)
      session.reload
      return if session.stream_key.present?
    rescue Youtube::SetupBroadcast::Error => e
      raise unless e.message.include?("temporaneamente") || e.message.include?("rate limit")
      raise if i >= max_attempts - 1

      wait = 60 + (i * 15)
      puts "  rate limit Google, attendo #{wait}s (tentativo #{i + 2}/#{max_attempts})..."
      sleep wait
    end
  end

  def resolve_e2e_session!
    if ENV["SESSION_ID"].present?
      return StreamSession.find(ENV["SESSION_ID"])
    end

    user = User.find_by(email: ENV.fetch("E2E_USER_EMAIL", "coach@matchlivetv.test"))
    abort "Utente E2E non trovato" unless user

    team = user.teams.first || Team.first
    match = team.matches.order(scheduled_at: :desc).first
    match ||= team.matches.create!(
      opponent_name: "E2E Test #{Time.current.strftime("%H:%M")}",
      scheduled_at: 5.minutes.from_now,
      location: "Test automatico"
    )

    Sessions::Create.new(
      user: user,
      match: match,
      params: { platform: "youtube", privacy_status: "unlisted", youtube_channel: "platform" }
    ).call
  end

  def wait_for(label, timeout_sec)
    deadline = Time.now + timeout_sec
    loop do
      if yield
        puts "  ✓ #{label}"
        return true
      end
      break if Time.now >= deadline

      sleep 3
      print "."
    end
    puts "\n  ✗ timeout: #{label}"
    false
  end

  def push_test_rtmp!(session, duration:)
    intake = "rtmp://mediamtx:1935/#{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", duration.to_s,
      "-c:v", "libx264", "-preset", "veryfast", "-pix_fmt", "yuv420p", "-b:v", "2500k",
      "-c:a", "aac", "-b:a", "128k",
      "-f", "flv", intake
    ]
    pid = Process.spawn(*cmd)
    unless Process.wait2(pid)[1].success?
      warn "RTMP test inviato con warning (exit non zero)"
    end
  end

  def ingest_status(session)
    client = Youtube::BroadcastService.new(session.match.team).send(:authorized_client)
    b = client.list_live_broadcasts("status,contentDetails", id: session.youtube_broadcast_id).items&.first
    return "n/a" unless b

    sid = b.content_details.bound_stream_id
    stream = client.list_live_streams("status", id: sid).items&.first
    "#{b.status.life_cycle_status} / ingest=#{stream&.status&.stream_status}"
  rescue StandardError => e
    e.message
  end

  def relay_summary(session)
    pid = Redis.new(url: ENV.fetch("REDIS_URL", "redis://localhost:6379/0"))
               .get(format(Streams::YoutubeRelay::REDIS_KEY, session.id))
    pid.present? ? "pid=#{pid} (sidekiq copy)" : "non attivo"
  end

  def youtube_ingest_active?(session)
    client = Youtube::BroadcastService.new(session.match.team).send(:authorized_client)
    b = client.list_live_broadcasts("status,contentDetails", id: session.youtube_broadcast_id).items&.first
    return false unless b

    sid = b.content_details.bound_stream_id
    stream = client.list_live_streams("status", id: sid).items&.first
    stream&.status&.stream_status == "active"
  rescue StandardError
    false
  end
end