From 05ef56c56dd23afc75cdbaf287bb7f9732a996ad Mon Sep 17 00:00:00 2001 From: Emiliano Frascaro Date: Fri, 28 Aug 2026 13:01:41 +0200 Subject: [PATCH] Aggiunge replay YouTube temporaneo, pull registrazioni dai CPX e snapshot ingest in admin. MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Così overflow Hetzner e VOD YouTube restano in archivio dopo lo spegnimento del nodo, e la colonna ingest non si svuota. Co-authored-by: Cursor --- .../controllers/admin/sessions_controller.rb | 11 +- .../api/v1/recordings_controller.rb | 5 +- .../controllers/api/v1/teams_controller.rb | 5 +- backend/app/helpers/admin_helper.rb | 10 +- .../recordings/clear_temporary_media_job.rb | 19 ++ .../app/jobs/recordings/post_process_job.rb | 24 +- .../recordings/purge_temporary_media_job.rb | 35 +++ .../recordings/verify_youtube_replay_job.rb | 47 ++++ backend/app/jobs/streams/autoscaler_job.rb | 12 + backend/app/models/recording.rb | 55 +++- backend/app/models/stream_session.rb | 17 +- .../recordings/clear_temporary_media.rb | 72 ++++++ .../services/recordings/finalize_session.rb | 71 ++++- .../recordings/pull_from_cloud_node.rb | 92 +++++++ .../app/services/recordings/storage_policy.rb | 36 +++ .../recordings/upload_from_session.rb | 62 ++++- .../recordings/verify_youtube_replay.rb | 86 +++++++ backend/app/services/sessions/create.rb | 26 +- backend/app/services/sessions/stop.rb | 2 +- backend/app/services/youtube/vod_status.rb | 85 ++++++ .../admin/sessions/_ingest_cell.html.erb | 11 +- .../app/views/admin/sessions/index.html.erb | 2 +- .../app/views/admin/sessions/show.html.erb | 20 +- backend/app/views/public/replay/show.html.erb | 50 ++-- .../views/recordings/_club_archive.html.erb | 6 +- backend/config/initializers/match_live_tv.rb | 18 ++ backend/config/locales/admin.de.yml | 1 + backend/config/locales/admin.en.yml | 1 + backend/config/locales/admin.es.yml | 1 + backend/config/locales/admin.fr.yml | 1 + backend/config/locales/admin.it.yml | 1 + ...d_temporary_replay_fields_to_recordings.rb | 15 ++ ..._add_ingest_snapshot_to_stream_sessions.rb | 51 ++++ backend/db/schema.rb | 10 +- backend/lib/tasks/recordings.rake | 33 +++ .../spec/models/stream_session_node_spec.rb | 31 +++ backend/spec/requests/admin/sessions_spec.rb | 27 ++ .../api/v1/public_replay_youtube_only_spec.rb | 60 +++++ .../v1/team_recordings_replay_source_spec.rb | 94 +++++++ .../recordings/finalize_session_spec.rb | 14 + .../recordings/pull_from_cloud_node_spec.rb | 79 ++++++ .../recordings/storage_policy_spec.rb | 40 +++ .../recordings/youtube_temp_replay_spec.rb | 243 ++++++++++++++++++ .../services/sessions/create_capacity_spec.rb | 50 ++++ docs/REPLAY_MODULE.md | 185 +++++++------ infra/.env.example | 6 + infra/.env.production.example | 5 + infra/scripts/install_production_cron.sh | 1 + infra/stream-node/cloud-init.yaml | 73 +++++- infra/stream-node/relay-agent.py | 72 +++++- .../match_live_tv/E2ECoverUploadTest.kt | 14 +- .../match_live_tv/data/api/ApiDtos.kt | 5 +- .../data/repository/SessionRepository.kt | 50 +++- scripts/deploy/bootstrap_stream_node.sh | 2 + 54 files changed, 1843 insertions(+), 201 deletions(-) create mode 100644 backend/app/jobs/recordings/clear_temporary_media_job.rb create mode 100644 backend/app/jobs/recordings/purge_temporary_media_job.rb create mode 100644 backend/app/jobs/recordings/verify_youtube_replay_job.rb create mode 100644 backend/app/services/recordings/clear_temporary_media.rb create mode 100644 backend/app/services/recordings/pull_from_cloud_node.rb create mode 100644 backend/app/services/recordings/storage_policy.rb create mode 100644 backend/app/services/recordings/verify_youtube_replay.rb create mode 100644 backend/app/services/youtube/vod_status.rb create mode 100644 backend/db/migrate/20260826193000_add_temporary_replay_fields_to_recordings.rb create mode 100644 backend/db/migrate/20260828124000_add_ingest_snapshot_to_stream_sessions.rb create mode 100644 backend/spec/requests/api/v1/public_replay_youtube_only_spec.rb create mode 100644 backend/spec/requests/api/v1/team_recordings_replay_source_spec.rb create mode 100644 backend/spec/services/recordings/pull_from_cloud_node_spec.rb create mode 100644 backend/spec/services/recordings/storage_policy_spec.rb create mode 100644 backend/spec/services/recordings/youtube_temp_replay_spec.rb create mode 100644 backend/spec/services/sessions/create_capacity_spec.rb diff --git a/backend/app/controllers/admin/sessions_controller.rb b/backend/app/controllers/admin/sessions_controller.rb index f3ba846..3ed3c89 100644 --- a/backend/app/controllers/admin/sessions_controller.rb +++ b/backend/app/controllers/admin/sessions_controller.rb @@ -79,14 +79,21 @@ module Admin if @filters[:status].present? && StreamSession::STATUSES.include?(@filters[:status]) scope = scope.where(stream_sessions: { status: @filters[:status] }) end - if @filters[:platform].present? && StreamSession::PLATFORMS.include?(@filters[:platform]) + if @filters[:platform].present? && StreamSession::LIVE_PLATFORMS.include?(@filters[:platform]) scope = scope.where(stream_sessions: { platform: @filters[:platform] }) end if @filters[:club_id].present? scope = scope.where(teams: { club_id: @filters[:club_id] }) end if @filters[:stream_node_id].present? - scope = scope.where(stream_sessions: { stream_node_id: @filters[:stream_node_id] }) + node = StreamNode.find_by(id: @filters[:stream_node_id]) + if node + scope = scope.where( + "stream_sessions.stream_node_id = :id OR stream_sessions.ingest_slug = :slug", + id: node.id, + slug: node.slug + ) + end end if (from_time = parse_filter_date(@filters[:from], end_of_day: false)) scope = scope.where( diff --git a/backend/app/controllers/api/v1/recordings_controller.rb b/backend/app/controllers/api/v1/recordings_controller.rb index f3da572..7c626f6 100644 --- a/backend/app/controllers/api/v1/recordings_controller.rb +++ b/backend/app/controllers/api/v1/recordings_controller.rb @@ -97,10 +97,11 @@ module Api replay_url: recording.replay_url, playback_url: recording.playback_stream_url, thumbnail_url: recording.thumbnail_url, - download_enabled: ent.phone_download_enabled?, + download_enabled: ent.phone_download_enabled? && recording.storage_key.present?, youtube_video_id: recording.youtube_video_id, youtube_watch_url: recording.youtube_watch_url, - youtube_publish_enabled: ent.premium_full? && ent.youtube_enabled?, + youtube_publish_enabled: ent.premium_full? && ent.youtube_enabled? && recording.storage_key.present? && recording.youtube_video_id.blank?, + replay_source: recording.replay_source, source_platform: recording.source_platform, source_platform_label: recording.source_platform_label, expires_at: recording.expires_at, diff --git a/backend/app/controllers/api/v1/teams_controller.rb b/backend/app/controllers/api/v1/teams_controller.rb index 7f91cab..9b890c7 100644 --- a/backend/app/controllers/api/v1/teams_controller.rb +++ b/backend/app/controllers/api/v1/teams_controller.rb @@ -171,12 +171,13 @@ module Api replay_url: recording.replay_url, playback_url: recording.playback_stream_url, thumbnail_url: recording.thumbnail_url, - download_enabled: ent.phone_download_enabled?, + download_enabled: ent.phone_download_enabled? && recording.storage_key.present?, view_count: recording.view_count, views_label: recording.views_label, youtube_video_id: recording.youtube_video_id, youtube_watch_url: recording.youtube_watch_url, - youtube_publish_enabled: ent.premium_full? && ent.youtube_enabled?, + youtube_publish_enabled: ent.premium_full? && ent.youtube_enabled? && recording.storage_key.present? && recording.youtube_video_id.blank?, + replay_source: recording.replay_source, source_platform: recording.source_platform, source_platform_label: recording.source_platform_label, expires_at: recording.expires_at, diff --git a/backend/app/helpers/admin_helper.rb b/backend/app/helpers/admin_helper.rb index 286f597..f9a32e9 100644 --- a/backend/app/helpers/admin_helper.rb +++ b/backend/app/helpers/admin_helper.rb @@ -30,8 +30,9 @@ module AdminHelper links end - def admin_session_ingest_badge_class(node) - case node.role + def admin_session_ingest_badge_class(role_or_node) + role = role_or_node.respond_to?(:role) ? role_or_node.role : role_or_node + case role.to_s when "home" then "badge--ingest-home" when "lab" then "badge--ingest-lab" when "cloud" then "badge--ingest-cloud" @@ -39,8 +40,9 @@ module AdminHelper end end - def admin_session_ingest_role_label(node) - I18n.t("admin.sessions.ingest.role.#{node.role}", default: node.role.to_s.humanize) + def admin_session_ingest_role_label(role_or_node) + role = role_or_node.respond_to?(:role) ? role_or_node.role : role_or_node + I18n.t("admin.sessions.ingest.role.#{role}", default: role.to_s.humanize) end def admin_session_status_badge_class(status) diff --git a/backend/app/jobs/recordings/clear_temporary_media_job.rb b/backend/app/jobs/recordings/clear_temporary_media_job.rb new file mode 100644 index 0000000..2b2bcc0 --- /dev/null +++ b/backend/app/jobs/recordings/clear_temporary_media_job.rb @@ -0,0 +1,19 @@ +# frozen_string_literal: true + +module Recordings + class ClearTemporaryMediaJob + include Sidekiq::Job + sidekiq_options retry: 3, queue: "default" + + def perform(recording_id, reason = "verified") + recording = Recording.find_by(id: recording_id) + return unless recording + + Recordings::ClearTemporaryMedia.new( + recording, + reason: reason.to_sym, + force: reason.to_s == "verified" + ).call + end + end +end diff --git a/backend/app/jobs/recordings/post_process_job.rb b/backend/app/jobs/recordings/post_process_job.rb index b09ec46..0ca3100 100644 --- a/backend/app/jobs/recordings/post_process_job.rb +++ b/backend/app/jobs/recordings/post_process_job.rb @@ -1,3 +1,5 @@ +# frozen_string_literal: true + module Recordings class PostProcessJob include Sidekiq::Job @@ -7,10 +9,30 @@ module Recordings recording = Recording.find_by(id: recording_id) return unless recording&.ready? - Recordings::NotifyReady.new(recording).call + begin + Recordings::NotifyReady.new(recording).call + rescue StandardError => e + # SMTP/ntfy giù non deve bloccare VerifyYoutubeReplay né il resto del post-process. + Rails.logger.warn( + "[Recordings::PostProcessJob] notify_failed recording=#{recording.id} " \ + "#{e.class}: #{e.message}" + ) + end + if recording.temporary_storage? + grace = MatchLiveTv.youtube_replay_verify_grace_secs + Rails.logger.info( + "[Recordings::PostProcessJob] schedule VerifyYoutubeReplay " \ + "recording=#{recording.id} grace=#{grace}s" + ) + Recordings::VerifyYoutubeReplayJob.perform_in(grace.seconds, recording.id, 0) + return + end + + # Solo MatchLiveTV-only: eventuale re-upload manuale/flag legacy (non per live YouTube). return unless recording.auto_publish_youtube? return if recording.youtube_video_id.present? + return if recording.source_platform == "youtube" Recordings::PublishToYoutubeJob.perform_async(recording.id) end diff --git a/backend/app/jobs/recordings/purge_temporary_media_job.rb b/backend/app/jobs/recordings/purge_temporary_media_job.rb new file mode 100644 index 0000000..a26e450 --- /dev/null +++ b/backend/app/jobs/recordings/purge_temporary_media_job.rb @@ -0,0 +1,35 @@ +# frozen_string_literal: true + +module Recordings + # Safety net: elimina copie temp scadute (anche se verify non è mai riuscito). + class PurgeTemporaryMediaJob + include Sidekiq::Job + sidekiq_options retry: 1, queue: "default" + + def perform + scope = Recording.temporary_media_pending_purge + count = 0 + + scope.find_each do |recording| + unless recording.youtube_verified_at.present? + Rails.logger.warn( + "[Recordings::PurgeTemporaryMediaJob] anomaly_unverified_expiry " \ + "recording=#{recording.id} session=#{recording.stream_session_id} " \ + "youtube_video_id=#{recording.youtube_video_id.inspect} " \ + "temp_expires_at=#{recording.temp_expires_at.inspect}" + ) + end + + Recordings::ClearTemporaryMedia.new( + recording, + reason: :max_retention, + force: true + ).call + count += 1 + end + + Rails.logger.info("[Recordings::PurgeTemporaryMediaJob] processed=#{count}") + count + end + end +end diff --git a/backend/app/jobs/recordings/verify_youtube_replay_job.rb b/backend/app/jobs/recordings/verify_youtube_replay_job.rb new file mode 100644 index 0000000..fbb1015 --- /dev/null +++ b/backend/app/jobs/recordings/verify_youtube_replay_job.rb @@ -0,0 +1,47 @@ +# frozen_string_literal: true + +module Recordings + class VerifyYoutubeReplayJob + include Sidekiq::Job + sidekiq_options retry: 0, queue: "default" + + def perform(recording_id, attempt = 0) + recording = Recording.find_by(id: recording_id) + return unless recording&.temporary_storage? + return if recording.deleted? + + attempt = attempt.to_i + max = MatchLiveTv.youtube_replay_verify_max_attempts + result = Recordings::VerifyYoutubeReplay.new(recording).call + + if result.ok? + Recordings::ClearTemporaryMediaJob.perform_async(recording.id, "verified") + return + end + + if result.retriable? && attempt + 1 < max + delay = backoff_secs(attempt) + Rails.logger.info( + "[Recordings::VerifyYoutubeReplayJob] retry recording=#{recording.id} " \ + "attempt=#{attempt + 1}/#{max} in=#{delay}s msg=#{result.message}" + ) + self.class.perform_in(delay.seconds, recording.id, attempt + 1) + return + end + + Rails.logger.warn( + "[Recordings::VerifyYoutubeReplayJob] give_up recording=#{recording.id} " \ + "attempts=#{attempt + 1} msg=#{result.message} " \ + "temp_expires_at=#{recording.temp_expires_at.inspect}" + ) + end + + private + + def backoff_secs(attempt) + base = MatchLiveTv.youtube_replay_verify_base_interval_secs + # 300, 600, 1200, ... capped at 1h + [base * (2**attempt), 3600].min + end + end +end diff --git a/backend/app/jobs/streams/autoscaler_job.rb b/backend/app/jobs/streams/autoscaler_job.rb index a0e44c4..3f73521 100644 --- a/backend/app/jobs/streams/autoscaler_job.rb +++ b/backend/app/jobs/streams/autoscaler_job.rb @@ -8,6 +8,8 @@ module Streams INTERVAL_SECS = ENV.fetch("STREAM_AUTOSCALE_INTERVAL_SECS", "60").to_i REDIS_CHAIN_KEY = "streams:autoscaler:chain" + KICK_DEBOUNCE_KEY = "streams:autoscaler:kick" + KICK_DEBOUNCE_SECS = ENV.fetch("STREAM_AUTOSCALE_KICK_DEBOUNCE_SECS", "5").to_i def self.ensure_chain return unless redis @@ -17,6 +19,16 @@ module Streams perform_in(INTERVAL_SECS) end + # Kick immediato (es. NoCapacity / soft-free bassi). Debounce anti-flood Sidekiq. + def self.kick! + return false unless Streams::Autoscaler.enabled? + return false unless redis + return false unless redis.set(KICK_DEBOUNCE_KEY, "1", nx: true, ex: KICK_DEBOUNCE_SECS) + + perform_async + true + end + def self.redis @redis ||= Redis.new(url: ENV.fetch("REDIS_URL", "redis://localhost:6379/0")) rescue Redis::CannotConnectError diff --git a/backend/app/models/recording.rb b/backend/app/models/recording.rb index c9ae0cd..a1b1df8 100644 --- a/backend/app/models/recording.rb +++ b/backend/app/models/recording.rb @@ -2,6 +2,8 @@ class Recording < ApplicationRecord STATUSES = %w[processing ready expired failed].freeze PRIVACY_STATUSES = %w[public unlisted].freeze STORAGE_BACKENDS = %w[local s3].freeze + STORAGE_POLICIES = %w[temporary retained none].freeze + REPLAY_SOURCES = %w[youtube matchlivetv none].freeze belongs_to :stream_session belongs_to :team @@ -9,6 +11,7 @@ class Recording < ApplicationRecord validates :status, inclusion: { in: STATUSES } validates :privacy_status, inclusion: { in: PRIVACY_STATUSES } validates :storage_backend, inclusion: { in: STORAGE_BACKENDS } + validates :storage_policy, inclusion: { in: STORAGE_POLICIES } scope :not_deleted, -> { where(deleted_at: nil) } scope :ready, lambda { @@ -23,7 +26,16 @@ class Recording < ApplicationRecord ready.where(expires_at: ..days.days.from_now) } scope :expired_pending_purge, lambda { - not_deleted.where(status: %w[ready failed]).where("expires_at IS NOT NULL AND expires_at <= ?", Time.current) + not_deleted + .where(storage_policy: "retained") + .where(status: %w[ready failed]) + .where("expires_at IS NOT NULL AND expires_at <= ?", Time.current) + } + scope :temporary_media_pending_purge, lambda { + not_deleted + .where(storage_policy: "temporary") + .where(local_media_purged_at: nil) + .where("temp_expires_at IS NOT NULL AND temp_expires_at <= ?", Time.current) } scope :search_replays, lambda { |query| q = query.to_s.strip @@ -46,19 +58,30 @@ class Recording < ApplicationRecord end def playback_stream_url - return nil unless ready? && stream_session_id.present? + return nil unless ready? && storage_key.present? && stream_session_id.present? "#{MatchLiveTv.app_public_url.chomp('/')}/replay/#{stream_session_id}/stream" end def thumbnail_url + return youtube_thumbnail_url if thumbnail_storage_key.blank? && youtube_video_id.present? return nil unless thumbnail_storage_key.present? && stream_session_id.present? "#{MatchLiveTv.app_public_url.chomp('/')}/replay/#{stream_session_id}/thumbnail" end + def youtube_thumbnail_url + return nil if youtube_video_id.blank? + return nil if youtube_video_id.to_s.start_with?("mock_") + + meta_url = metadata.is_a?(Hash) ? metadata.dig("youtube", "thumbnail_url") : nil + return meta_url if meta_url.present? + + "https://i.ytimg.com/vi/#{youtube_video_id}/hqdefault.jpg" + end + def download_api_path - return nil unless ready? + return nil unless ready? && storage_key.present? "/api/v1/recordings/#{id}/download" end @@ -70,6 +93,30 @@ class Recording < ApplicationRecord "https://www.youtube.com/watch?v=#{youtube_video_id}" end + def temporary_storage? + storage_policy == "temporary" + end + + def retained_storage? + storage_policy == "retained" + end + + def local_media_purged? + local_media_purged_at.present? + end + + # Sorgente fisica del player in archivio (non lo storage policy). + def replay_source + return "youtube" if youtube_watch_url.present? || (ready? && youtube_video_id.present?) + return "matchlivetv" if ready? && storage_key.present? + + "none" + end + + def available_in_archive? + ready? && (youtube_watch_url.present? || youtube_video_id.present? || storage_key.present?) + end + def ready? status == "ready" && !deleted? && (expires_at.nil? || expires_at.future?) end @@ -154,7 +201,7 @@ class Recording < ApplicationRecord end def playable_on_site? - ready? && storage_key.present? + available_in_archive? end def days_until_expiry diff --git a/backend/app/models/stream_session.rb b/backend/app/models/stream_session.rb index 2373bcf..1181dcc 100644 --- a/backend/app/models/stream_session.rb +++ b/backend/app/models/stream_session.rb @@ -1,7 +1,8 @@ class StreamSession < ApplicationRecord include AASM - PLATFORMS = %w[matchlivetv youtube facebook twitch].freeze + LIVE_PLATFORMS = %w[matchlivetv youtube].freeze + PLATFORMS = (LIVE_PLATFORMS + %w[facebook twitch]).freeze STATUSES = %w[idle connecting live reconnecting paused ended error].freeze PRIVACY_STATUSES = %w[public unlisted private].freeze @@ -20,6 +21,7 @@ class StreamSession < ApplicationRecord before_validation :normalize_privacy_status before_validation :ensure_publish_token, on: :create + before_validation :snapshot_ingest_from_node, if: -> { stream_node.present? } scope :broadcasting, -> { where(status: %w[live connecting reconnecting paused]) } scope :publicly_listed, -> { where(privacy_status: "public") } @@ -74,6 +76,14 @@ class StreamSession < ApplicationRecord end end + def ingest_slug_display + stream_node&.slug.presence || ingest_slug + end + + def ingest_role_display + stream_node&.role.presence || ingest_role + end + def rtmp_ingest_url # RootEncoder richiede rtmp://host:port/app/stream (due segmenti). # MediaMTX path = live/match_{uuid} (no ?token= nel path). @@ -208,6 +218,11 @@ class StreamSession < ApplicationRecord self.privacy_status = "unlisted" if privacy_status == "private" end + def snapshot_ingest_from_node + self.ingest_slug = stream_node.slug + self.ingest_role = stream_node.role + end + def record_ended_timestamps! now = Time.current update!(ended_at: now) if ended_at.nil? diff --git a/backend/app/services/recordings/clear_temporary_media.rb b/backend/app/services/recordings/clear_temporary_media.rb new file mode 100644 index 0000000..7f36e0c --- /dev/null +++ b/backend/app/services/recordings/clear_temporary_media.rb @@ -0,0 +1,72 @@ +# frozen_string_literal: true + +module Recordings + # Rimuove solo l'MP4 temporaneo su object storage. Non soft-delete, non tocca YouTube. + class ClearTemporaryMedia + def initialize(recording, reason: :verified, force: false) + @recording = recording + @reason = reason + @force = force + end + + def call + unless @recording.temporary_storage? + Rails.logger.info( + "[Recordings::ClearTemporaryMedia] skip recording=#{@recording.id} not_temporary" + ) + return @recording + end + + if @recording.local_media_purged? + Rails.logger.info( + "[Recordings::ClearTemporaryMedia] already_purged recording=#{@recording.id}" + ) + return @recording + end + + unless allowed? + Rails.logger.info( + "[Recordings::ClearTemporaryMedia] skip recording=#{@recording.id} " \ + "reason=not_verified_and_not_expired force=#{@force}" + ) + return @recording + end + + delete_video_object! + @recording.update!( + storage_key: nil, + byte_size: nil, + local_media_purged_at: Time.current + ) + + Rails.logger.info( + "[Recordings::ClearTemporaryMedia] purged recording=#{@recording.id} " \ + "reason=#{@reason} youtube_verified=#{@recording.youtube_verified_at.present?} " \ + "youtube_video_id=#{@recording.youtube_video_id.inspect}" + ) + @recording + end + + private + + def allowed? + return true if @force + return true if @recording.youtube_verified_at.present? + return true if @recording.temp_expires_at.present? && @recording.temp_expires_at <= Time.current + + false + end + + def delete_video_object! + key = @recording.storage_key + return if key.blank? + + Recordings::Storage.new.delete(key: key) + rescue Recordings::Storage::Error => e + # File già assente → ok (idempotente) + Rails.logger.info( + "[Recordings::ClearTemporaryMedia] storage delete recording=#{@recording.id}: #{e.message}" + ) + end + end +end diff --git a/backend/app/services/recordings/finalize_session.rb b/backend/app/services/recordings/finalize_session.rb index 2e7bb7a..6a5dea6 100644 --- a/backend/app/services/recordings/finalize_session.rb +++ b/backend/app/services/recordings/finalize_session.rb @@ -1,3 +1,5 @@ +# frozen_string_literal: true + module Recordings class FinalizeSession def initialize(session) @@ -5,29 +7,69 @@ module Recordings end def call - team = @session.match.team - return unless team.entitlements.can_create_recordings? + policy = Recordings::StoragePolicy.call(@session) + return if policy == Recordings::StoragePolicy::NONE - retention_days = team.entitlements.recording_retention_days - expires_at = retention_days.positive? ? retention_days.days.from_now : nil + team = @session.match.team + attrs = attributes_for(policy, team) recording = Recording.find_or_initialize_by(stream_session: @session) - recording.assign_attributes( + if skip_reinitialize?(recording) + Rails.logger.info( + "[Recordings::FinalizeSession] skip already-finalized " \ + "recording=#{recording.id} status=#{recording.status}" + ) + return recording + end + recording.assign_attributes(attrs) + recording.save! + + Rails.logger.info( + "[Recordings::FinalizeSession] session=#{@session.id} recording=#{recording.id} " \ + "storage_policy=#{policy} expires_at=#{recording.expires_at.inspect} " \ + "temp_expires_at=#{recording.temp_expires_at.inspect}" + ) + recording + end + + private + + def skip_reinitialize?(recording) + return false unless recording.persisted? + + recording.ready? || + recording.status == "processing" || + recording.storage_key.present? + end + + def attributes_for(policy, team) + { team: team, status: "processing", title: default_title, privacy_status: privacy_from_session, storage_path: @session.mediamtx_path_name, recorded_at: @session.ended_at || Time.current, - expires_at: expires_at, + storage_policy: policy, + expires_at: archive_expires_at(policy, team), + temp_expires_at: temp_expires_at_for(policy), error_message: nil, - metadata: initial_metadata(team) - ) - recording.save! - recording + metadata: initial_metadata(policy, team) + } end - private + def archive_expires_at(policy, team) + return nil if policy == Recordings::StoragePolicy::TEMPORARY + + retention_days = team.entitlements.recording_retention_days + retention_days.positive? ? retention_days.days.from_now : nil + end + + def temp_expires_at_for(policy) + return nil unless policy == Recordings::StoragePolicy::TEMPORARY + + MatchLiveTv.youtube_temp_replay_retention_hours.hours.from_now + end def default_title match = @session.match @@ -38,12 +80,13 @@ module Recordings @session.privacy_status == "public" ? "public" : "unlisted" end - def initial_metadata(team) - ent = team.entitlements + def initial_metadata(policy, team) { "source_platform" => @session.platform, "session_privacy" => @session.privacy_status, - "auto_publish_youtube" => ent.premium_full? && ent.youtube_enabled? && @session.platform == "youtube", + # Re-upload automatico disabilitato: le live YouTube usano VerifyYoutubeReplay. + "auto_publish_youtube" => false, + "storage_policy" => policy, "ai" => {} } end diff --git a/backend/app/services/recordings/pull_from_cloud_node.rb b/backend/app/services/recordings/pull_from_cloud_node.rb new file mode 100644 index 0000000..7cd2546 --- /dev/null +++ b/backend/app/services/recordings/pull_from_cloud_node.rb @@ -0,0 +1,92 @@ +# frozen_string_literal: true + +require "cgi" +require "fileutils" +require "net/http" +require "uri" + +module Recordings + # Recupera i segmenti MediaMTX dal disco del CPX (agent :9100) verso un tmpdir locale. + class PullFromCloudNode + class Error < StandardError; end + + def initialize(session) + @session = session + end + + def applicable? + node = @session.stream_node + node.present? && node.role == "cloud" && agent_url.present? + end + + # @return [String, nil] directory con i file, o nil se il nodo non è cloud / 404 + def fetch + return unless applicable? + + dest = Dir.mktmpdir("mltv-cpx-rec-") + uri = recordings_uri + http = Net::HTTP.new(uri.host, uri.port) + http.open_timeout = 5 + http.read_timeout = ENV.fetch("STREAM_NODE_RECORDINGS_PULL_TIMEOUT", "180").to_i + req = Net::HTTP::Get.new(uri) + req["Authorization"] = "Bearer #{agent_secret}" if agent_secret.present? + res = http.request(req) + if res.is_a?(Net::HTTPNotFound) + FileUtils.remove_entry(dest) + return nil + end + unless res.is_a?(Net::HTTPSuccess) && res.body.present? + FileUtils.remove_entry(dest) + raise Error, "agent GET recordings HTTP #{res.code} #{res.body.to_s.truncate(200)}" + end + + tar_path = File.join(dest, "recordings.tar.gz") + File.binwrite(tar_path, res.body) + unpack!(tar_path, dest) + FileUtils.rm_f(tar_path) + dest + rescue StandardError + FileUtils.remove_entry(dest) if dest && Dir.exist?(dest) + raise + end + + def cleanup_remote! + return unless applicable? + + uri = recordings_uri + http = Net::HTTP.new(uri.host, uri.port) + http.open_timeout = 5 + http.read_timeout = 15 + req = Net::HTTP::Delete.new(uri) + req["Authorization"] = "Bearer #{agent_secret}" if agent_secret.present? + res = http.request(req) + return if res.is_a?(Net::HTTPSuccess) || res.is_a?(Net::HTTPNotFound) + + Rails.logger.warn( + "[Recordings::PullFromCloudNode] delete HTTP #{res.code} session=#{@session.id}" + ) + rescue StandardError => e + Rails.logger.warn("[Recordings::PullFromCloudNode] delete #{e.class}: #{e.message}") + end + + private + + def recordings_uri + path = @session.mediamtx_path_name.to_s + URI.parse("#{agent_url.chomp('/')}/recordings/#{CGI.escape(path)}") + end + + def agent_url + ENV["STREAM_NODE_RELAY_AGENT_URL"].presence || @session.stream_node&.relay_agent_url + end + + def agent_secret + ENV["STREAM_NODE_AGENT_SECRET"].presence || "mediamtx_webhook_dev_secret" + end + + def unpack!(tar_path, dest) + ok = system("tar", "-xzf", tar_path, "-C", dest, out: File::NULL, err: File::NULL) + raise Error, "tar extract failed" unless ok + end + end +end diff --git a/backend/app/services/recordings/storage_policy.rb b/backend/app/services/recordings/storage_policy.rb new file mode 100644 index 0000000..3f2fc95 --- /dev/null +++ b/backend/app/services/recordings/storage_policy.rb @@ -0,0 +1,36 @@ +# frozen_string_literal: true + +module Recordings + # Decisione centralizzata: temporary (YouTube) | retained (HLS) | none. + class StoragePolicy + TEMPORARY = "temporary" + RETAINED = "retained" + NONE = "none" + POLICIES = [TEMPORARY, RETAINED, NONE].freeze + + def self.call(session) + new(session).call + end + + def initialize(session) + @session = session + end + + def call + team = @session.match.team + ent = team.entitlements + + return NONE unless ent.can_create_recordings? + + if youtube_destination? + TEMPORARY + else + RETAINED + end + end + + def youtube_destination? + @session.platform.to_s == "youtube" + end + end +end diff --git a/backend/app/services/recordings/upload_from_session.rb b/backend/app/services/recordings/upload_from_session.rb index 34c6eb0..5460cc8 100644 --- a/backend/app/services/recordings/upload_from_session.rb +++ b/backend/app/services/recordings/upload_from_session.rb @@ -9,8 +9,9 @@ module Recordings def call recording = Recording.find_by(stream_session: @session) return unless recording&.status == "processing" + return if recording.storage_key.present? - source_files = local_source_files + source_files = collect_source_files if source_files.empty? fail_recording!(recording, "Nessun file di registrazione trovato") cleanup_mediamtx_path @@ -35,23 +36,52 @@ module Recordings error_message: nil ) + Rails.logger.info( + "[Recordings::UploadFromSession] ready recording=#{recording.id} " \ + "storage_policy=#{recording.storage_policy} key=#{storage_key} bytes=#{byte_size}" + ) + cleanup_local_sources(source_files, merged_path) + cleanup_cloud_pull! + cleanup_remote_recordings! cleanup_mediamtx_path Recordings::PostProcessJob.perform_async(recording.id) recording - rescue Error, Recordings::Storage::Error => e + rescue Error, Recordings::Storage::Error, Recordings::PullFromCloudNode::Error => e recording = Recording.find_by(stream_session: @session) fail_recording!(recording, e.message) if recording raise ensure FileUtils.rm_f(@merged_temp_path) if @merged_temp_path && File.exist?(@merged_temp_path) + cleanup_cloud_pull! unless recording_ready? end private + def collect_source_files + files = local_source_files + return files if files.any? + + puller = Recordings::PullFromCloudNode.new(@session) + return [] unless puller.applicable? + + 3.times do |i| + cleanup_cloud_pull! + @cloud_pull_dir = puller.fetch + files = scan_recording_files(@cloud_pull_dir) + return files if files.any? + + sleep 2 if i < 2 + end + [] + end + def local_source_files - base = File.join(MatchLiveTv.recordings_local_path, @session.mediamtx_path_name) - return [] unless Dir.exist?(base) + scan_recording_files(File.join(MatchLiveTv.recordings_local_path, @session.mediamtx_path_name)) + end + + def scan_recording_files(base) + return [] if base.blank? || !Dir.exist?(base) Dir.glob(File.join(base, "**", "*")) .select { |path| File.file?(path) && path.match?(/\.(mp4|fmp4|m4s|ts)$/i) } @@ -83,7 +113,11 @@ module Recordings end def object_key(recording) - "teams/#{recording.team_id}/sessions/#{@session.id}/replay.mp4" + if recording.temporary_storage? + "temporary_replays/teams/#{recording.team_id}/sessions/#{@session.id}/replay.mp4" + else + "teams/#{recording.team_id}/sessions/#{@session.id}/replay.mp4" + end end def cleanup_local_sources(source_files, merged_path) @@ -100,6 +134,24 @@ module Recordings Rails.logger.warn("[Recordings::UploadFromSession] delete_path: #{e.message}") end + def cleanup_cloud_pull! + return if @cloud_pull_dir.blank? || !Dir.exist?(@cloud_pull_dir) + + FileUtils.remove_entry(@cloud_pull_dir) + @cloud_pull_dir = nil + rescue StandardError + nil + end + + def cleanup_remote_recordings! + Recordings::PullFromCloudNode.new(@session).cleanup_remote! + end + + def recording_ready? + rec = Recording.find_by(stream_session: @session) + rec&.status == "ready" + end + def fail_recording!(recording, message) recording.update!(status: "failed", error_message: message) cleanup_mediamtx_path diff --git a/backend/app/services/recordings/verify_youtube_replay.rb b/backend/app/services/recordings/verify_youtube_replay.rb new file mode 100644 index 0000000..252e40d --- /dev/null +++ b/backend/app/services/recordings/verify_youtube_replay.rb @@ -0,0 +1,86 @@ +# frozen_string_literal: true + +module Recordings + # Collega la live YouTube (broadcast_id) al VOD e aggiorna metadata recording. + class VerifyYoutubeReplay + class Error < StandardError; end + + Result = Struct.new(:status, :message, keyword_init: true) do + def ok? + status == :ok + end + + def retriable? + status == :pending + end + end + + def initialize(recording) + @recording = recording + end + + def call + unless @recording.temporary_storage? + return Result.new(status: :skipped, message: "not_temporary") + end + + if @recording.youtube_verified_at.present? && @recording.youtube_video_id.present? + return Result.new(status: :ok, message: "already_verified") + end + + session = @recording.stream_session + broadcast_id = session&.youtube_broadcast_id + if broadcast_id.blank? + Rails.logger.warn( + "[Recordings::VerifyYoutubeReplay] missing broadcast_id recording=#{@recording.id}" + ) + return Result.new(status: :failed, message: "missing_broadcast_id") + end + + info = Youtube::VodStatus.new(@recording.team, channel: "team").fetch(broadcast_id) + unless info.ready + Rails.logger.info( + "[Recordings::VerifyYoutubeReplay] not_ready recording=#{@recording.id} " \ + "video=#{broadcast_id} upload_status=#{info.upload_status.inspect}" + ) + return Result.new(status: :pending, message: "vod_not_ready") + end + + apply_verified!(info) + Rails.logger.info( + "[Recordings::VerifyYoutubeReplay] ok recording=#{@recording.id} " \ + "youtube_video_id=#{info.video_id}" + ) + Result.new(status: :ok, message: "verified") + rescue Youtube::VodStatus::Error => e + Rails.logger.warn( + "[Recordings::VerifyYoutubeReplay] api_error recording=#{@recording.id}: #{e.message}" + ) + Result.new(status: :pending, message: e.message) + end + + private + + def apply_verified!(info) + meta = @recording.metadata.is_a?(Hash) ? @recording.metadata.deep_dup : {} + yt = meta.fetch("youtube", {}).merge( + "broadcast_id" => @recording.stream_session.youtube_broadcast_id, + "thumbnail_url" => info.thumbnail_url, + "privacy_status" => info.privacy_status, + "upload_status" => info.upload_status, + "verified_via" => "live_broadcast" + ).compact + + attrs = { + youtube_video_id: info.video_id, + youtube_verified_at: Time.current, + youtube_published_at: @recording.youtube_published_at || Time.current, + metadata: meta.merge("youtube" => yt) + } + attrs[:duration_secs] = info.duration_secs if info.duration_secs.to_i.positive? + attrs[:title] = info.title if info.title.present? && @recording.title.blank? + + @recording.update!(attrs) + end + end +end diff --git a/backend/app/services/sessions/create.rb b/backend/app/services/sessions/create.rb index 2ad8476..b778387 100644 --- a/backend/app/services/sessions/create.rb +++ b/backend/app/services/sessions/create.rb @@ -55,13 +55,15 @@ module Sessions ) end + kick_autoscaler_if_soft_limit! + if session.platform == "youtube" YoutubeBroadcastSetupJob.perform_later(session.id, youtube_channel) end session rescue Streams::NodeRegistry::NoCapacityError => e - raise Teams::EntitlementError.new(e.message, code: "stream_capacity_exhausted") + raise_no_capacity!(e) rescue Faraday::ConnectionFailed, Faraday::TimeoutError => e raise Streams::IngestUnavailableError, "Ingest temporaneamente non disponibile (#{e.class})" rescue Mediamtx::Client::Error => e @@ -73,6 +75,28 @@ module Sessions private + def raise_no_capacity!(error) + if Streams::Autoscaler.enabled? + Streams::AutoscalerJob.kick! + raise Teams::EntitlementError.new( + "Capacità streaming in espansione. Riprova tra poco.", + code: "stream_capacity_scaling" + ) + end + + raise Teams::EntitlementError.new(error.message, code: "stream_capacity_exhausted") + end + + def kick_autoscaler_if_soft_limit! + return unless Streams::Autoscaler.enabled? + + free = Streams::Autoscaler.metrics[:free_slots].to_i + return if free > Streams::Autoscaler.soft_free_slots + + Streams::AutoscalerJob.kick! + end + + def assert_youtube_channel!(youtube_channel) resolver = Youtube::CredentialResolver.new(@match.team, channel: youtube_channel) if resolver.resolve.blank? diff --git a/backend/app/services/sessions/stop.rb b/backend/app/services/sessions/stop.rb index 96c9a74..453bdc5 100644 --- a/backend/app/services/sessions/stop.rb +++ b/backend/app/services/sessions/stop.rb @@ -12,7 +12,7 @@ module Sessions complete_youtube_broadcast! if @session.youtube_broadcast_id.present? recording = Recordings::FinalizeSession.new(@session).call - Recordings::UploadJob.perform_async(@session.id) if recording + Recordings::UploadJob.perform_async(@session.id) if recording&.status == "processing" remove_mediamtx_paths! log_event("ended") diff --git a/backend/app/services/youtube/vod_status.rb b/backend/app/services/youtube/vod_status.rb new file mode 100644 index 0000000..1d70a89 --- /dev/null +++ b/backend/app/services/youtube/vod_status.rb @@ -0,0 +1,85 @@ +# frozen_string_literal: true + +module Youtube + # Stato VOD dopo una live (broadcast_id ≈ video_id su YouTube). + class VodStatus + class Error < StandardError; end + + Result = Struct.new( + :ready, + :video_id, + :title, + :duration_secs, + :thumbnail_url, + :privacy_status, + :upload_status, + keyword_init: true + ) + + def initialize(team, channel: "team") + @team = team + @channel = channel + end + + def fetch(video_id) + raise Error, "video_id mancante" if video_id.blank? + + if mock_or_unconfigured?(video_id) + return Result.new( + ready: true, + video_id: video_id, + title: nil, + duration_secs: nil, + thumbnail_url: nil, + privacy_status: "unlisted", + upload_status: "processed" + ) + end + + client = authorized_client + item = client.list_videos("snippet,contentDetails,status", id: video_id).items&.first + return Result.new(ready: false, video_id: video_id) if item.blank? + + upload_status = item.status&.upload_status.to_s + ready = upload_status.in?(%w[processed uploaded]) || + (item.snippet.present? && upload_status != "deleted" && upload_status != "rejected" && upload_status != "failed") + + Result.new( + ready: ready, + video_id: item.id, + title: item.snippet&.title, + duration_secs: parse_duration(item.content_details&.duration), + thumbnail_url: item.snippet&.thumbnails&.high&.url || item.snippet&.thumbnails&.default&.url, + privacy_status: item.status&.privacy_status, + upload_status: upload_status + ) + rescue Google::Apis::Error => e + raise Error, e.message + end + + private + + def mock_or_unconfigured?(video_id) + video_id.to_s.start_with?("mock_") || + ENV["YOUTUBE_CLIENT_ID"].blank? || + CredentialResolver.new(@team, channel: @channel).resolve.blank? + end + + def authorized_client + credential = CredentialResolver.new(@team, channel: @channel).resolve + raise Error, "Credenziali YouTube non disponibili" if credential.blank? + + OauthRefresh.new(credential).apply!(Google::Apis::YoutubeV3::YouTubeService.new) + end + + def parse_duration(iso) + return nil if iso.blank? + + # PT1H2M3S + match = iso.match(/\APT(?:(\d+)H)?(?:(\d+)M)?(?:(\d+)S)?\z/) + return nil unless match + + match[1].to_i * 3600 + match[2].to_i * 60 + match[3].to_i + end + end +end diff --git a/backend/app/views/admin/sessions/_ingest_cell.html.erb b/backend/app/views/admin/sessions/_ingest_cell.html.erb index 47482f6..bd8c7fa 100644 --- a/backend/app/views/admin/sessions/_ingest_cell.html.erb +++ b/backend/app/views/admin/sessions/_ingest_cell.html.erb @@ -1,8 +1,11 @@ -<% node = session.stream_node %> -<% if node %> +<% slug = session.ingest_slug_display %> +<% role = session.ingest_role_display %> +<% if slug.present? %>
- <%= node.slug %> - <%= admin_session_ingest_role_label(node) %> + <%= slug %> + <% if role.present? %> + <%= admin_session_ingest_role_label(role) %> + <% end %>
<% else %> <%= t("admin.sessions.ingest.none") %> diff --git a/backend/app/views/admin/sessions/index.html.erb b/backend/app/views/admin/sessions/index.html.erb index da18444..72fe3c5 100644 --- a/backend/app/views/admin/sessions/index.html.erb +++ b/backend/app/views/admin/sessions/index.html.erb @@ -26,7 +26,7 @@ <%= t("admin.sessions.index.filters.platform") %> <%= select_tag :platform, options_for_select( - [[t("admin.sessions.index.filters.any"), ""]] + StreamSession::PLATFORMS.map { |p| [p, p] }, + [[t("admin.sessions.index.filters.any"), ""]] + StreamSession::LIVE_PLATFORMS.map { |p| [p, p] }, @filters[:platform] ) %> diff --git a/backend/app/views/admin/sessions/show.html.erb b/backend/app/views/admin/sessions/show.html.erb index c8871b0..70eedfa 100644 --- a/backend/app/views/admin/sessions/show.html.erb +++ b/backend/app/views/admin/sessions/show.html.erb @@ -162,12 +162,20 @@
<%= t("admin.sessions.show.fields.node") %>
- <% if @session.stream_node %> - <%= @session.stream_node.slug %> - - <%= admin_session_ingest_role_label(@session.stream_node) %> - - (<%= @session.stream_node.provider %>) + <% slug = @session.ingest_slug_display %> + <% role = @session.ingest_role_display %> + <% if slug.present? %> + <%= slug %> + <% if role.present? %> + + <%= admin_session_ingest_role_label(role) %> + + <% end %> + <% if @session.stream_node %> + (<%= @session.stream_node.provider %>) + <% else %> + <%= t("admin.sessions.ingest.decommissioned") %> + <% end %> <% else %> <%= t("admin.sessions.ingest.none") %> <% end %> diff --git a/backend/app/views/public/replay/show.html.erb b/backend/app/views/public/replay/show.html.erb index 717c743..e6ef7b7 100644 --- a/backend/app/views/public/replay/show.html.erb +++ b/backend/app/views/public/replay/show.html.erb @@ -14,6 +14,33 @@

<%= t("replay.show.processing_title") %>

<%= t("replay.show.processing_body") %>

+ <% elsif @recording.ready? && @recording.replay_source == "youtube" && @recording.youtube_video_id.present? && !@recording.youtube_video_id.to_s.start_with?("mock_") %> +
+ + <%= render "public/live/player_overlays", + match: @match, + session: @session, + stream_closed: true, + on_air: false, + badge_label: t("replay.show.badge_label"), + badge_class: "badge-ended" %> +
+

+ <%= l_local(@recording.recorded_at_or_fallback) %> + · <%= t("replay.show.meta_line_duration", value: @recording.duration_label) %> + · <%= @recording.views_label %> + <% if @recording.source_platform_label != "—" %> + · <%= @recording.source_platform_label %> + <% end %> +

+

+ <%= t("replay.show.youtube_only_body") %> + <%= link_to t("replay.show.youtube_link"), @recording.youtube_watch_url, target: "_blank", rel: "noopener" %> +

<% elsif @recording.ready? && @recording.storage_key.present? %>