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? if Mediamtx::PublisherOnline.active?(session) && session.platform == "youtube" Mediamtx::PublisherSync.schedule_youtube_pipeline!(session) end Streams::YoutubeRelay.stop(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 }) end end end