MatchLiveTv/backend/app/services/webhooks/mediamtx_handler.rb

module Webhooks
  class MediamtxHandler
    def initialize(event:, session_id:, metadata: {})
      @event = event
      @session_id = session_id
      @metadata = metadata
    end

    def call
      session = StreamSession.find_by(id: @session_id)
      return unless session

      case @event
      when "connect"
        handle_connect(session)
      when "disconnect"
        handle_disconnect(session)
      when "ready"
        handle_ready(session)
      end
    end

    private

    def handle_connect(session)
      return if session.paused?

      Streams::OverlayRelay.ensure_publishing!(session)
      Streams::YoutubeRelay.ensure_publishing!(session) if session.platform == "youtube"

      return if session.live? && session.stream_events.where(event_type: "connected").exists?

      if session.reconnecting?
        session.reconnect! if session.may_reconnect?
        enable_recording(session)
        cancel_timeout(session)
        log(session, "reconnected")
        broadcast(session, "reconnected")
      elsif session.connecting? || session.idle?
        session.go_live! if session.may_go_live?
        enable_recording(session)
        log(session, "connected")
        broadcast(session, "connected")
      end
    end

    def handle_disconnect(session)
      disable_recording(session)
      return if session.paused?
      return unless session.live? || session.connecting?

      if session.connecting?
        session.update!(status: "reconnecting")
      elsif session.may_lose_connection?
        session.lose_connection!
      end
      log(session, "disconnected", @metadata)
      broadcast(session, "disconnected")
      schedule_timeout(session)
    end

    def handle_ready(session)
      log(session, "connected", { ready: true })
    end

    def enable_recording(session)
      return unless session.match.team.entitlements.recording_enabled_for_mediamtx?

      Mediamtx::Client.new.set_path_recording(session, enabled: true)
    rescue Mediamtx::Client::Error => e
      Rails.logger.warn("[MediamtxHandler] enable recording: #{e.message}")
    end

    def disable_recording(session)
      Mediamtx::Client.new.set_path_recording(session, enabled: false)
    rescue Mediamtx::Client::Error => e
      Rails.logger.warn("[MediamtxHandler] disable recording: #{e.message}")
    end

    def schedule_timeout(session)
      job = DisconnectionTimeoutJob.perform_in(
        MatchLiveTv.reconnect_timeout_seconds.seconds,
        session.id
      )
      session.update!(timeout_job_id: job)
    end

    def cancel_timeout(session)
      return if session.timeout_job_id.blank?

      Sidekiq::ScheduledSet.new.find_job(session.timeout_job_id)&.delete
      session.update!(timeout_job_id: nil)
    end

    def log(session, type, meta = {})
      return if duplicate_event?(session, type, meta)

      session.stream_events.create!(
        event_type: type,
        metadata: meta.merge(source: "mediamtx"),
        occurred_at: Time.current
      )
    end

    def duplicate_event?(session, type, meta)
      session.stream_events
             .where(event_type: type)
             .where("occurred_at > ?", 2.seconds.ago)
             .where(metadata: meta.merge(source: "mediamtx"))
             .exists?
    end

    def broadcast(session, event)
      SessionChannel.broadcast_message(session, { type: "stream_event", event: event })
      Streams::Overlay::Refresh.call(session)
    end
  end
end