Initial commit: monorepo Match Live TV.
Rails API, app Flutter, infrastruttura Docker/MediaMTX, sito marketing e documentazione di deploy. Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
94
backend/app/services/webhooks/mediamtx_handler.rb
Normal file
94
backend/app/services/webhooks/mediamtx_handler.rb
Normal file
@@ -0,0 +1,94 @@
|
||||
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.live? && session.stream_events.where(event_type: "connected").exists?
|
||||
|
||||
if session.reconnecting?
|
||||
session.reconnect! if session.may_reconnect?
|
||||
cancel_timeout(session)
|
||||
log(session, "reconnected")
|
||||
broadcast(session, "reconnected")
|
||||
elsif session.connecting? || session.idle?
|
||||
session.go_live! if session.may_go_live?
|
||||
log(session, "connected")
|
||||
broadcast(session, "connected")
|
||||
end
|
||||
end
|
||||
|
||||
def handle_disconnect(session)
|
||||
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 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
|
||||
14
backend/app/services/webhooks/signature.rb
Normal file
14
backend/app/services/webhooks/signature.rb
Normal file
@@ -0,0 +1,14 @@
|
||||
module Webhooks
|
||||
class Signature
|
||||
def self.valid?(payload, signature)
|
||||
return false if signature.blank?
|
||||
|
||||
expected = OpenSSL::HMAC.hexdigest(
|
||||
"SHA256",
|
||||
MatchLiveTv.mediamtx_webhook_secret,
|
||||
payload
|
||||
)
|
||||
ActiveSupport::SecurityUtils.secure_compare(expected, signature)
|
||||
end
|
||||
end
|
||||
end
|
||||
Reference in New Issue
Block a user