From 0a6e7dda4d56f0e7b76eae6bbfc7dd189d40fffb Mon Sep 17 00:00:00 2001 From: zijiren233 Date: Sat, 20 Jun 2026 12:29:23 +0800 Subject: [PATCH] chore --- .../src/grpc/client_service/room/streaming.rs | 2 +- .../providers/playback_provider/transport.rs | 5 +- synctv-api/src/http/room/playback.rs | 4 +- synctv-api/src/impls/admin/batch.rs | 2 +- synctv-api/src/impls/admin/reports.rs | 7 +- synctv-api/src/impls/admin/reviews.rs | 34 ++---- synctv-api/src/impls/client/file_download.rs | 107 ++++++++++-------- synctv-api/src/impls/messaging/codec.rs | 3 +- synctv-api/src/impls/providers/common.rs | 2 +- synctv-cluster/src/discovery/node_registry.rs | 2 - .../src/discovery/static_discovery.rs | 14 ++- synctv-core/src/provider/live_helpers.rs | 12 +- synctv-core/src/repository/review.rs | 11 +- synctv-core/src/service/cleanup.rs | 10 +- synctv-core/src/service/cleanup_ops.rs | 5 +- synctv-core/src/service/db_maintenance.rs | 8 +- synctv-core/src/service/providers_manager.rs | 8 +- synctv-core/src/service/room/entries.rs | 18 ++- synctv-core/src/service/room/member_admin.rs | 17 ++- synctv-core/src/service/room/outbox.rs | 6 +- synctv-core/src/service/room/settings.rs | 6 +- synctv-livestream/src/util.rs | 14 +-- synctv-management/src/mapping.rs | 6 +- synctv-management/src/service.rs | 4 +- synctv-media-providers/src/alist/client.rs | 10 +- synctv-media-providers/src/bilibili/mod.rs | 20 ++-- synctv-media-providers/src/emby/client.rs | 9 +- .../src/slice_cache/backend/file_loader.rs | 1 - .../src/slice_cache/backend/file_tests.rs | 20 ++-- synctv-proxy/src/slice_cache/store.rs | 6 +- .../src/sync/connection_manager.rs | 97 ++++++++-------- synctv-realtime/src/sync/redis_pubsub.rs | 34 ++++-- synctv/src/app.rs | 3 +- synctv/src/cli.rs | 8 +- 34 files changed, 263 insertions(+), 252 deletions(-) diff --git a/synctv-api/src/grpc/client_service/room/streaming.rs b/synctv-api/src/grpc/client_service/room/streaming.rs index c7dfe5ab..66f81d6c 100644 --- a/synctv-api/src/grpc/client_service/room/streaming.rs +++ b/synctv-api/src/grpc/client_service/room/streaming.rs @@ -7,7 +7,6 @@ use super::super::{ invalid_argument_status, map_api_error, map_message_stream_join_error, map_message_stream_membership_error, realtime_room_access_error, ClientServiceImpl, }; -use crate::impls::ApiError; use super::RoomService; use crate::grpc::client_service::streaming::{ watch_chat_events_event, watch_chat_pin_events_event, watch_playback_event, @@ -18,6 +17,7 @@ use crate::impls::messaging::{ GuestRealtimeIdentity, MessageConcurrencyConfig, MessageSender, RealtimePrincipal, StreamMessage, StreamMessageHandler, StreamMessageHandlerConfig, StreamMessageHandlerRuntime, }; +use crate::impls::ApiError; use crate::impls::EndpointRateLimitCategory; use synctv_proto::client::*; diff --git a/synctv-api/src/http/providers/playback_provider/transport.rs b/synctv-api/src/http/providers/playback_provider/transport.rs index 5058a31d..27309c20 100644 --- a/synctv-api/src/http/providers/playback_provider/transport.rs +++ b/synctv-api/src/http/providers/playback_provider/transport.rs @@ -1,10 +1,7 @@ use axum::{ body::Body, http::{header, HeaderMap, HeaderValue, Method, StatusCode}, - response::{ - sse::Event, - Response, - }, + response::{sse::Event, Response}, }; use futures::{Stream, StreamExt}; use std::convert::Infallible; diff --git a/synctv-api/src/http/room/playback.rs b/synctv-api/src/http/room/playback.rs index 17c0d25b..2a66061e 100644 --- a/synctv-api/src/http/room/playback.rs +++ b/synctv-api/src/http/room/playback.rs @@ -220,8 +220,8 @@ pub async fn watch_bilibili_live_danmaku( }, ) .await?; - let stream = - stream.map(crate::http::providers::playback_provider::transport::bilibili_danmaku_sse_event); + let stream = stream + .map(crate::http::providers::playback_provider::transport::bilibili_danmaku_sse_event); Ok(Sse::new(stream).keep_alive(KeepAlive::default())) } diff --git a/synctv-api/src/impls/admin/batch.rs b/synctv-api/src/impls/admin/batch.rs index 91f183eb..d7c312d0 100644 --- a/synctv-api/src/impls/admin/batch.rs +++ b/synctv-api/src/impls/admin/batch.rs @@ -1,7 +1,7 @@ use synctv_core::models::{UserId, UserRole}; use super::{ - check_role_hierarchy, list_owned_room_ids, BatchResultsAccumulator, AdminApiImpl, ApiError, + check_role_hierarchy, list_owned_room_ids, AdminApiImpl, ApiError, BatchResultsAccumulator, RequestContext, }; diff --git a/synctv-api/src/impls/admin/reports.rs b/synctv-api/src/impls/admin/reports.rs index 7355ca9a..798de1bb 100644 --- a/synctv-api/src/impls/admin/reports.rs +++ b/synctv-api/src/impls/admin/reports.rs @@ -24,11 +24,8 @@ impl AdminApiImpl { "reporter_user_id", &self.public_id_codec, )?; - let room_id = crate::impls::parse_optional_id_param( - &req.room_id, - "room_id", - &self.public_id_codec, - )?; + let room_id = + crate::impls::parse_optional_id_param(&req.room_id, "room_id", &self.public_id_codec)?; let target_room_id = crate::impls::parse_optional_id_param( &req.target_room_id, "target_room_id", diff --git a/synctv-api/src/impls/admin/reviews.rs b/synctv-api/src/impls/admin/reviews.rs index 1977be3c..74ef1caa 100644 --- a/synctv-api/src/impls/admin/reviews.rs +++ b/synctv-api/src/impls/admin/reviews.rs @@ -8,9 +8,9 @@ use synctv_core::{ use super::{ ban_row_to_proto, i64_to_i32_api, normalize_non_empty_filter, pagination_limit_offset_i64, - proto_review_status_filter, room_creation_review_row_to_proto, - room_join_review_row_to_proto, user_registration_review_row_to_proto, AdminApiImpl, ApiError, - RequestContext, LOCAL_MANAGEMENT_ACTOR_USER_ID, + proto_review_status_filter, room_creation_review_row_to_proto, room_join_review_row_to_proto, + user_registration_review_row_to_proto, AdminApiImpl, ApiError, RequestContext, + LOCAL_MANAGEMENT_ACTOR_USER_ID, }; impl AdminApiImpl { @@ -217,16 +217,10 @@ impl AdminApiImpl { self.require_admin_actor(admin_user_id).await?; let (limit, offset) = pagination_limit_offset_i64(req.page, req.page_size, "room join review")?; - let room_id_filter = crate::impls::parse_optional_id_param( - &req.room_id, - "room_id", - &self.public_id_codec, - )?; - let user_id_filter = crate::impls::parse_optional_id_param( - &req.user_id, - "user_id", - &self.public_id_codec, - )?; + let room_id_filter = + crate::impls::parse_optional_id_param(&req.room_id, "room_id", &self.public_id_codec)?; + let user_id_filter = + crate::impls::parse_optional_id_param(&req.user_id, "user_id", &self.public_id_codec)?; let page = self .review_service @@ -316,16 +310,10 @@ impl AdminApiImpl { self.require_admin_actor(admin_user_id).await?; let (limit, offset) = pagination_limit_offset_i64(req.page, req.page_size, "ban record")?; - let user_id_filter = crate::impls::parse_optional_id_param( - &req.user_id, - "user_id", - &self.public_id_codec, - )?; - let room_id_filter = crate::impls::parse_optional_id_param( - &req.room_id, - "room_id", - &self.public_id_codec, - )?; + let user_id_filter = + crate::impls::parse_optional_id_param(&req.user_id, "user_id", &self.public_id_codec)?; + let room_id_filter = + crate::impls::parse_optional_id_param(&req.room_id, "room_id", &self.public_id_codec)?; let page = self .ban_record_service diff --git a/synctv-api/src/impls/client/file_download.rs b/synctv-api/src/impls/client/file_download.rs index 233dbe5a..f8bab696 100644 --- a/synctv-api/src/impls/client/file_download.rs +++ b/synctv-api/src/impls/client/file_download.rs @@ -49,15 +49,18 @@ pub(crate) fn avatar_chunk_stream( ) -> impl futures::Stream> + Send + 'static { - generic_chunk_stream(download, |mime_type, content_manifest_sha256, data, content_range, total_size_bytes| { - synctv_proto::client::UserAvatarObjectResponse { - mime_type, - content_manifest_sha256, - data, - content_range, - total_size_bytes, - } - }) + generic_chunk_stream( + download, + |mime_type, content_manifest_sha256, data, content_range, total_size_bytes| { + synctv_proto::client::UserAvatarObjectResponse { + mime_type, + content_manifest_sha256, + data, + content_range, + total_size_bytes, + } + }, + ) } pub(crate) fn chat_attachment_chunk_stream( @@ -66,16 +69,19 @@ pub(crate) fn chat_attachment_chunk_stream( ) -> impl futures::Stream> + Send + 'static { - generic_chunk_stream(download, move |mime_type, content_manifest_sha256, data, content_range, total_size_bytes| { - synctv_proto::client::ChatAttachmentObjectResponse { - room_id: room_id.clone(), - mime_type, - content_manifest_sha256, - data, - content_range, - total_size_bytes, - } - }) + generic_chunk_stream( + download, + move |mime_type, content_manifest_sha256, data, content_range, total_size_bytes| { + synctv_proto::client::ChatAttachmentObjectResponse { + room_id: room_id.clone(), + mime_type, + content_manifest_sha256, + data, + content_range, + total_size_bytes, + } + }, + ) } pub(crate) fn media_cover_chunk_stream( @@ -83,15 +89,18 @@ pub(crate) fn media_cover_chunk_stream( ) -> impl futures::Stream> + Send + 'static { - generic_chunk_stream(download, |mime_type, content_manifest_sha256, data, content_range, total_size_bytes| { - synctv_proto::client::MediaCoverObjectResponse { - mime_type, - content_manifest_sha256, - data, - content_range, - total_size_bytes, - } - }) + generic_chunk_stream( + download, + |mime_type, content_manifest_sha256, data, content_range, total_size_bytes| { + synctv_proto::client::MediaCoverObjectResponse { + mime_type, + content_manifest_sha256, + data, + content_range, + total_size_bytes, + } + }, + ) } pub(crate) fn room_cover_chunk_stream( @@ -99,15 +108,18 @@ pub(crate) fn room_cover_chunk_stream( ) -> impl futures::Stream> + Send + 'static { - generic_chunk_stream(download, |mime_type, content_manifest_sha256, data, content_range, total_size_bytes| { - synctv_proto::client::RoomCoverObjectResponse { - mime_type, - content_manifest_sha256, - data, - content_range, - total_size_bytes, - } - }) + generic_chunk_stream( + download, + |mime_type, content_manifest_sha256, data, content_range, total_size_bytes| { + synctv_proto::client::RoomCoverObjectResponse { + mime_type, + content_manifest_sha256, + data, + content_range, + total_size_bytes, + } + }, + ) } pub(crate) fn playlist_cover_chunk_stream( @@ -115,15 +127,18 @@ pub(crate) fn playlist_cover_chunk_stream( ) -> impl futures::Stream> + Send + 'static { - generic_chunk_stream(download, |mime_type, content_manifest_sha256, data, content_range, total_size_bytes| { - synctv_proto::client::PlaylistCoverObjectResponse { - mime_type, - content_manifest_sha256, - data, - content_range, - total_size_bytes, - } - }) + generic_chunk_stream( + download, + |mime_type, content_manifest_sha256, data, content_range, total_size_bytes| { + synctv_proto::client::PlaylistCoverObjectResponse { + mime_type, + content_manifest_sha256, + data, + content_range, + total_size_bytes, + } + }, + ) } #[cfg(test)] diff --git a/synctv-api/src/impls/messaging/codec.rs b/synctv-api/src/impls/messaging/codec.rs index 22a14fbc..0e7ed74a 100644 --- a/synctv-api/src/impls/messaging/codec.rs +++ b/synctv-api/src/impls/messaging/codec.rs @@ -196,7 +196,8 @@ pub(crate) fn room_member_event_to_proto( } => { let room_id = encode_room(*room_id)?; let user_id = encode_user(*target_user_id)?; - let validated_username = required_realtime_text(target_username, "target username", 50)?; + let validated_username = + required_realtime_text(target_username, "target username", 50)?; let member = synctv_proto::common::RoomMember { room_id: room_id.clone(), user_id: user_id.clone(), diff --git a/synctv-api/src/impls/providers/common.rs b/synctv-api/src/impls/providers/common.rs index 1b555056..812c5719 100644 --- a/synctv-api/src/impls/providers/common.rs +++ b/synctv-api/src/impls/providers/common.rs @@ -4,8 +4,8 @@ use synctv_core::models::{ ProviderInstance, UserProviderCredential, }; use synctv_core::models::{SortDirection as CoreSortDirection, UserId}; -use synctv_core::provider::{ExecutionControl, ProviderAccessService}; use synctv_core::provider::ProviderError; +use synctv_core::provider::{ExecutionControl, ProviderAccessService}; use synctv_core::repository::UserProviderCredentialRepository; use synctv_core::service::{ AuditEventParams, AuditService, ProvidersManager, RemoteProviderManager, UserService, diff --git a/synctv-cluster/src/discovery/node_registry.rs b/synctv-cluster/src/discovery/node_registry.rs index 535ce2c1..27ee5cda 100644 --- a/synctv-cluster/src/discovery/node_registry.rs +++ b/synctv-cluster/src/discovery/node_registry.rs @@ -182,7 +182,6 @@ static REGISTER_REMOTE_NODE_SCRIPT: LazyLock = LazyLock::new(|| { ) }); - /// Create a failsafe circuit breaker for Redis operations. /// /// Opens after 3 consecutive failures. Uses exponential backoff starting at @@ -1390,7 +1389,6 @@ impl NodeRegistry { Ok(()) } - /// Unregister a remote node with epoch validation /// /// Uses the same atomic Lua script pattern as `unregister()` to validate diff --git a/synctv-cluster/src/discovery/static_discovery.rs b/synctv-cluster/src/discovery/static_discovery.rs index be8ff13d..60ea55f8 100644 --- a/synctv-cluster/src/discovery/static_discovery.rs +++ b/synctv-cluster/src/discovery/static_discovery.rs @@ -113,11 +113,8 @@ impl StaticDiscovery { } Some( - NodeInfo::new( - discovered.node_id.clone(), - normalized_api_address, - ) - .with_epoch(discovered.epoch), + NodeInfo::new(discovered.node_id.clone(), normalized_api_address) + .with_epoch(discovered.epoch), ) } @@ -264,7 +261,12 @@ impl StaticDiscovery { connect_timeout: Duration, cluster_secret: &str, ) -> Option { - super::probe_node_identity(normalized_api_address, connect_timeout.as_secs(), cluster_secret).await + super::probe_node_identity( + normalized_api_address, + connect_timeout.as_secs(), + cluster_secret, + ) + .await } } diff --git a/synctv-core/src/provider/live_helpers.rs b/synctv-core/src/provider/live_helpers.rs index 81bbd227..009fe0b8 100644 --- a/synctv-core/src/provider/live_helpers.rs +++ b/synctv-core/src/provider/live_helpers.rs @@ -61,18 +61,10 @@ pub(super) fn live_ids_from_metadata( context_label: &str, ) -> Result<(RoomId, MediaId), ProviderError> { let room_id = metadata_typed_id(versioned, "room_id", |room_id| { - super::playback_transport::parse_playback_room_id( - public_id_codec, - room_id, - context_label, - ) + super::playback_transport::parse_playback_room_id(public_id_codec, room_id, context_label) })?; let media_id = metadata_typed_id(versioned, "media_id", |media_id| { - super::playback_transport::parse_playback_media_id( - public_id_codec, - media_id, - context_label, - ) + super::playback_transport::parse_playback_media_id(public_id_codec, media_id, context_label) })?; Ok((room_id, media_id)) } diff --git a/synctv-core/src/repository/review.rs b/synctv-core/src/repository/review.rs index 4d049df1..5a67e8f0 100644 --- a/synctv-core/src/repository/review.rs +++ b/synctv-core/src/repository/review.rs @@ -5,8 +5,8 @@ use crate::models::{ oauth2_provider_type_name_from_code, ReviewRequestId, ReviewStatus, RoomId, SignupMethod, UserId, }; -use crate::repository::query_builder::escape_ilike; use crate::repository::pools::RepoPools; +use crate::repository::query_builder::escape_ilike; use crate::{Error, Result}; fn count_value(value: Option, query_description: &str) -> Result { @@ -284,8 +284,13 @@ impl ReviewRepository { reviewed_by: Option, reason: &str, ) -> Result { - Self::reject_user_registration_with_executor(self.pools.primary(), request_id, reviewed_by, reason) - .await + Self::reject_user_registration_with_executor( + self.pools.primary(), + request_id, + reviewed_by, + reason, + ) + .await } pub async fn approve_user_registration_with_executor<'e, E>( diff --git a/synctv-core/src/service/cleanup.rs b/synctv-core/src/service/cleanup.rs index 526f8789..e34d5db6 100644 --- a/synctv-core/src/service/cleanup.rs +++ b/synctv-core/src/service/cleanup.rs @@ -23,13 +23,11 @@ use std::sync::Arc; use tokio_util::sync::CancellationToken; use tracing::{info, warn}; -use super::{cleanup_ops, FileStorageCleanupOrigin, FileStorageService, LeaderCheck, SettingsRegistry}; -use crate::service::partitioning::u32_to_i32; -use crate::{ - models::ChatAttachment, - repository::FileStorageRepository, - InternalExt, Result, +use super::{ + cleanup_ops, FileStorageCleanupOrigin, FileStorageService, LeaderCheck, SettingsRegistry, }; +use crate::service::partitioning::u32_to_i32; +use crate::{models::ChatAttachment, repository::FileStorageRepository, InternalExt, Result}; const DEFAULT_ROOM_RESOURCE_EVENT_RETENTION_SECONDS: u64 = 30 * 24 * 60 * 60; diff --git a/synctv-core/src/service/cleanup_ops.rs b/synctv-core/src/service/cleanup_ops.rs index fb8a1bab..7237a8ad 100644 --- a/synctv-core/src/service/cleanup_ops.rs +++ b/synctv-core/src/service/cleanup_ops.rs @@ -17,10 +17,7 @@ use crate::{InternalExt, Result}; /// Delete expired media provider credentials with a buffer that prevents races /// with in-flight refreshes. Returns the number of rows deleted. -pub(super) async fn delete_expired_credentials( - pool: &PgPool, - buffer_hours: u32, -) -> Result { +pub(super) async fn delete_expired_credentials(pool: &PgPool, buffer_hours: u32) -> Result { if buffer_hours == 0 { return Ok(0); } diff --git a/synctv-core/src/service/db_maintenance.rs b/synctv-core/src/service/db_maintenance.rs index fa0cfed5..72107366 100644 --- a/synctv-core/src/service/db_maintenance.rs +++ b/synctv-core/src/service/db_maintenance.rs @@ -133,9 +133,11 @@ impl DatabaseMaintenanceService { /// Delete old notifications using the shared cleanup retention settings. pub async fn run_cleanup_notifications(&self) -> crate::Result<()> { - let read_deleted = - cleanup_ops::delete_old_read_notifications(&self.pool, self.config.notification_retention_days) - .await?; + let read_deleted = cleanup_ops::delete_old_read_notifications( + &self.pool, + self.config.notification_retention_days, + ) + .await?; let expired_deleted = cleanup_ops::delete_expired_notifications( &self.pool, self.config.notification_max_retention_days, diff --git a/synctv-core/src/service/providers_manager.rs b/synctv-core/src/service/providers_manager.rs index 87791b9e..9c1a43b6 100644 --- a/synctv-core/src/service/providers_manager.rs +++ b/synctv-core/src/service/providers_manager.rs @@ -204,14 +204,14 @@ impl ProvidersManager { let ssrf_guard = self.ssrf_guard.clone(); // Helper to select client manager based on config - let select_client_manager = |config: &Value, ssrf_guard: &synctv_common::ssrf::SsrfGuard| { - provider_http_client_from_config(config, ssrf_guard) - .map(|client_opt| { + let select_client_manager = + |config: &Value, ssrf_guard: &synctv_common::ssrf::SsrfGuard| { + provider_http_client_from_config(config, ssrf_guard).map(|client_opt| { client_opt.map(|client| { Arc::new(ProviderClientManager::new_with_provider_http_client(client)) }) }) - }; + }; // Alist factory let default_client_manager_alist = Arc::clone(&default_client_manager); diff --git a/synctv-core/src/service/room/entries.rs b/synctv-core/src/service/room/entries.rs index 3c574b74..a117dcc6 100644 --- a/synctv-core/src/service/room/entries.rs +++ b/synctv-core/src/service/room/entries.rs @@ -256,7 +256,11 @@ impl RoomService { &actor_username, *media_id, ); - super::outbox::log_if_no_local_subscribers(subscriber_count, &room_id, "Media removed"); + super::outbox::log_if_no_local_subscribers( + subscriber_count, + &room_id, + "Media removed", + ); } for playlist_id in &impact.deleted_playlist_ids { let subscriber_count = self.notification_service.notify_playlist_deleted( @@ -265,7 +269,11 @@ impl RoomService { &actor_username, *playlist_id, ); - super::outbox::log_if_no_local_subscribers(subscriber_count, &room_id, "Playlist deleted"); + super::outbox::log_if_no_local_subscribers( + subscriber_count, + &room_id, + "Playlist deleted", + ); } } @@ -567,7 +575,11 @@ impl RoomService { &actor_username, *playlist_id, ); - super::outbox::log_if_no_local_subscribers(subscriber_count, &room_id, "Playlist deleted after clear_playlist"); + super::outbox::log_if_no_local_subscribers( + subscriber_count, + &room_id, + "Playlist deleted after clear_playlist", + ); } clear_playlist_result_from_impact(impact) diff --git a/synctv-core/src/service/room/member_admin.rs b/synctv-core/src/service/room/member_admin.rs index c087c025..0173cf57 100644 --- a/synctv-core/src/service/room/member_admin.rs +++ b/synctv-core/src/service/room/member_admin.rs @@ -1283,7 +1283,10 @@ impl RoomService { params: PermissionWriteParams<'_>, ) -> Result { let (added, removed) = if params.effective_is_admin { - (params.admin_added_permissions, params.admin_removed_permissions) + ( + params.admin_added_permissions, + params.admin_removed_permissions, + ) } else { (params.added_permissions, params.removed_permissions) }; @@ -1373,7 +1376,8 @@ impl RoomService { .invalidate_committed_member_write_cache(&snapshot.room_id, &snapshot.target_user_id) .await; if snapshot.role_changed { - self.notify_room_settings_invalidation(&snapshot.room_id).await; + self.notify_room_settings_invalidation(&snapshot.room_id) + .await; } Ok(()) } @@ -1391,7 +1395,14 @@ impl RoomService { outbox_event_factory: Option<&RealtimeOutboxPermissionChangedEventFactory>, ) -> Result { let snapshot = self - .permission_changed_snapshot_tx(tx, room_id, target_user_id, actor_id, updated, role_changed) + .permission_changed_snapshot_tx( + tx, + room_id, + target_user_id, + actor_id, + updated, + role_changed, + ) .await?; self.insert_permission_changed_outbox_tx(tx, &snapshot, outbox_event_factory) .await?; diff --git a/synctv-core/src/service/room/outbox.rs b/synctv-core/src/service/room/outbox.rs index 718e2971..41d855d5 100644 --- a/synctv-core/src/service/room/outbox.rs +++ b/synctv-core/src/service/room/outbox.rs @@ -18,7 +18,11 @@ use crate::{ }; /// Helper: Log if a notification had no local subscribers (Finding 10) -pub(super) fn log_if_no_local_subscribers(subscriber_count: usize, room_id: &RoomId, event_label: &str) { +pub(super) fn log_if_no_local_subscribers( + subscriber_count: usize, + room_id: &RoomId, + event_label: &str, +) { if subscriber_count == 0 { tracing::debug!( room_id = %room_id, diff --git a/synctv-core/src/service/room/settings.rs b/synctv-core/src/service/room/settings.rs index 64447f47..bfc60f15 100644 --- a/synctv-core/src/service/room/settings.rs +++ b/synctv-core/src/service/room/settings.rs @@ -588,7 +588,11 @@ impl RoomService { settings_json.clone(), version, ); - super::outbox::log_if_no_local_subscribers(subscriber_count, room_id, "Room settings updated"); + super::outbox::log_if_no_local_subscribers( + subscriber_count, + room_id, + "Room settings updated", + ); Ok(crate::cache::RoomSettingsSnapshot { settings: updated_settings.clone(), diff --git a/synctv-livestream/src/util.rs b/synctv-livestream/src/util.rs index 3a02c963..e226fb28 100644 --- a/synctv-livestream/src/util.rs +++ b/synctv-livestream/src/util.rs @@ -9,11 +9,7 @@ const MAX_HLS_SEGMENT_NAME_LEN: usize = 256; const MAX_HLS_SEGMENT_URL_PART_LEN: usize = 2048; /// Shared validation logic for stream identifiers and segment names. -fn validate_identifier_common( - value: &str, - field: &str, - max_len: usize, -) -> anyhow::Result<()> { +fn validate_identifier_common(value: &str, field: &str, max_len: usize) -> anyhow::Result<()> { if value.is_empty() { return Err(anyhow::anyhow!("{field} must not be empty")); } @@ -41,13 +37,7 @@ fn validate_identifier_common( /// Validate a stream identifier component before it is used in internal /// registry/cache keys. pub(crate) fn validate_stream_id_component(component: &str, field: &str) -> anyhow::Result<()> { - validate_identifier_common(component, field, MAX_STREAM_ID_COMPONENT_LEN)?; - if component.contains(':') { - return Err(anyhow::anyhow!( - "{field} must not contain ':' because stream keys use ':' as an internal delimiter" - )); - } - Ok(()) + validate_identifier_common(component, field, MAX_STREAM_ID_COMPONENT_LEN) } /// Validate canonical room/media identifiers used by livestream internals. diff --git a/synctv-management/src/mapping.rs b/synctv-management/src/mapping.rs index c39ee169..063153f7 100644 --- a/synctv-management/src/mapping.rs +++ b/synctv-management/src/mapping.rs @@ -13,14 +13,12 @@ pub(crate) fn map_user_role(role: i32) -> Result { } pub(crate) fn map_user_status(status: i32) -> Result { - common_proto::UserStatus::try_from(status) - .map_err(|_| invalid_enum_value("status", status))?; + common_proto::UserStatus::try_from(status).map_err(|_| invalid_enum_value("status", status))?; Ok(status) } pub(crate) fn map_room_status(status: i32) -> Result { - common_proto::RoomStatus::try_from(status) - .map_err(|_| invalid_enum_value("status", status))?; + common_proto::RoomStatus::try_from(status).map_err(|_| invalid_enum_value("status", status))?; Ok(status) } diff --git a/synctv-management/src/service.rs b/synctv-management/src/service.rs index 17c0c41a..fc0b964d 100644 --- a/synctv-management/src/service.rs +++ b/synctv-management/src/service.rs @@ -2247,9 +2247,7 @@ impl ManagementService for ManagementServiceImpl { ) .await .map_err(|e| map_api_error(&e))?; - Ok(Response::new( - client_proto::KickRoomStreamResponse {}, - )) + Ok(Response::new(client_proto::KickRoomStreamResponse {})) } async fn list_playlists( diff --git a/synctv-media-providers/src/alist/client.rs b/synctv-media-providers/src/alist/client.rs index c85bc4b5..e6d39570 100644 --- a/synctv-media-providers/src/alist/client.rs +++ b/synctv-media-providers/src/alist/client.rs @@ -213,9 +213,8 @@ impl AlistClient { let response = check_response(response).await?; let resp: AlistResp = json_with_limit(response).await?; check_alist_code(&resp)?; - resp.data.ok_or_else(|| { - AlistError::Parse(format!("Missing data in {ctx} response")) - }) + resp.data + .ok_or_else(|| AlistError::Parse(format!("Missing data in {ctx} response"))) } }) .await @@ -239,9 +238,8 @@ impl AlistClient { let response = check_response(response).await?; let resp: AlistResp = json_with_limit(response).await?; check_alist_code(&resp)?; - resp.data.ok_or_else(|| { - AlistError::Parse(format!("Missing data in {ctx} response")) - }) + resp.data + .ok_or_else(|| AlistError::Parse(format!("Missing data in {ctx} response"))) } }) .await diff --git a/synctv-media-providers/src/bilibili/mod.rs b/synctv-media-providers/src/bilibili/mod.rs index 698c010e..357c1c5c 100644 --- a/synctv-media-providers/src/bilibili/mod.rs +++ b/synctv-media-providers/src/bilibili/mod.rs @@ -20,15 +20,13 @@ pub use client::{ pub use error::BilibiliError; pub use service::{BilibiliInterface, BilibiliLiveDanmakuStream, BilibiliService}; pub use types::{ - CodecEntry, DashAudio, DashInfo, - DashPgcResp, DashPgcResult, DashVideo, DashVideoData, DashVideoResp, Dimension, DurlInfo, - Episode, EpisodeId, EpisodeInfo, EpisodePage, FormatEntry, GetLiveDanmuInfoResp, - GetLiveMasterInfoResp, GetLiveStreamResp, LiveDanmuData, LiveDanmuHost, LiveDurl, - LiveMasterData, LiveMasterInfo, LivePageData, LiveStreamData, NavData, NavResp, Owner, Page, - ParseLivePageResp, PgcUrlResp, PgcUrlResult, PlayUrlContainer, - PlayUrlInfoWrapper, PlayerV2Data, PlayerV2InfoResp, QrcodeData, - QrcodeResp, Quality, QualityDesc, RoomPlayInfoData, RoomPlayInfoResp, SeasonInfoResp, - SeasonResult, Section, SegmentBase, StreamEntry, SubtitleInfo, SubtitleItem, SupportFormat, - UgcSeason, UrlInfoEntry, VideoId, - VideoPageData, VideoPageInfoResp, VideoUrlData, VideoUrlResp, WbiImg, + CodecEntry, DashAudio, DashInfo, DashPgcResp, DashPgcResult, DashVideo, DashVideoData, + DashVideoResp, Dimension, DurlInfo, Episode, EpisodeId, EpisodeInfo, EpisodePage, FormatEntry, + GetLiveDanmuInfoResp, GetLiveMasterInfoResp, GetLiveStreamResp, LiveDanmuData, LiveDanmuHost, + LiveDurl, LiveMasterData, LiveMasterInfo, LivePageData, LiveStreamData, NavData, NavResp, + Owner, Page, ParseLivePageResp, PgcUrlResp, PgcUrlResult, PlayUrlContainer, PlayUrlInfoWrapper, + PlayerV2Data, PlayerV2InfoResp, QrcodeData, QrcodeResp, Quality, QualityDesc, RoomPlayInfoData, + RoomPlayInfoResp, SeasonInfoResp, SeasonResult, Section, SegmentBase, StreamEntry, + SubtitleInfo, SubtitleItem, SupportFormat, UgcSeason, UrlInfoEntry, VideoId, VideoPageData, + VideoPageInfoResp, VideoUrlData, VideoUrlResp, WbiImg, }; diff --git a/synctv-media-providers/src/emby/client.rs b/synctv-media-providers/src/emby/client.rs index c7c46371..860ed748 100644 --- a/synctv-media-providers/src/emby/client.rs +++ b/synctv-media-providers/src/emby/client.rs @@ -740,7 +740,8 @@ impl EmbyClient { let headers = self.build_headers()?; - self.send_post_no_response(&url, Some(&body), &headers).await + self.send_post_no_response(&url, Some(&body), &headers) + .await } /// Report playback stop to Emby server @@ -762,7 +763,8 @@ impl EmbyClient { let headers = self.build_headers()?; - self.send_post_no_response(&url, Some(&body), &headers).await + self.send_post_no_response(&url, Some(&body), &headers) + .await } /// Report playback progress to Emby server @@ -791,7 +793,8 @@ impl EmbyClient { let headers = self.build_headers()?; - self.send_post_no_response(&url, Some(&body), &headers).await + self.send_post_no_response(&url, Some(&body), &headers) + .await } /// Get host URL diff --git a/synctv-proxy/src/slice_cache/backend/file_loader.rs b/synctv-proxy/src/slice_cache/backend/file_loader.rs index 6368a833..535d1690 100644 --- a/synctv-proxy/src/slice_cache/backend/file_loader.rs +++ b/synctv-proxy/src/slice_cache/backend/file_loader.rs @@ -175,4 +175,3 @@ async fn load_cache_file( } } } - diff --git a/synctv-proxy/src/slice_cache/backend/file_tests.rs b/synctv-proxy/src/slice_cache/backend/file_tests.rs index b79d4bea..e0437a2a 100644 --- a/synctv-proxy/src/slice_cache/backend/file_tests.rs +++ b/synctv-proxy/src/slice_cache/backend/file_tests.rs @@ -278,12 +278,10 @@ async fn test_file_backend_evict_expired() -> TestResult { let evicted = backend.evict_expired().await; assert_eq!(evicted, 1); assert_eq!(backend.entry_count(), 1); - assert!( - backend - .get("fresh_key_0000000000000000000000000000000000000000000000000000000000") - .await - .is_some() - ); + assert!(backend + .get("fresh_key_0000000000000000000000000000000000000000000000000000000000") + .await + .is_some()); Ok(()) } @@ -344,12 +342,10 @@ async fn test_file_backend_evict_to_size() -> TestResult { backend.current_size() ); - assert!( - backend - .get("newest_key_0000000000000000000000000000000000000000000000000000000") - .await - .is_some() - ); + assert!(backend + .get("newest_key_0000000000000000000000000000000000000000000000000000000") + .await + .is_some()); Ok(()) } diff --git a/synctv-proxy/src/slice_cache/store.rs b/synctv-proxy/src/slice_cache/store.rs index 86717441..f091340f 100644 --- a/synctv-proxy/src/slice_cache/store.rs +++ b/synctv-proxy/src/slice_cache/store.rs @@ -218,11 +218,7 @@ impl SliceCache { /// Re-insert the still-cached slice data under a fresh segment TTL after a /// `304 Not Modified`, returning the data on success. - async fn refresh_cached_slice_ttl( - &self, - key: &str, - data: Bytes, - ) -> Result<(), anyhow::Error> { + async fn refresh_cached_slice_ttl(&self, key: &str, data: Bytes) -> Result<(), anyhow::Error> { let refreshed = StoredEntry::new(data, self.config.segment_ttl); self.backend .put(key, refreshed) diff --git a/synctv-realtime/src/sync/connection_manager.rs b/synctv-realtime/src/sync/connection_manager.rs index 7abbcbfb..bd09c8a0 100644 --- a/synctv-realtime/src/sync/connection_manager.rs +++ b/synctv-realtime/src/sync/connection_manager.rs @@ -1229,65 +1229,64 @@ impl ConnectionManager { let lifecycle_lock = self.connection_lifecycle_lock(connection_id); let lifecycle_guard = lifecycle_lock.lock().await; - let (transition, last_activity) = if let Some(mut conn) = - self.connections.get_mut(connection_id) - { - let current_room_id = conn.room_id; - if current_room_id.as_ref() == Some(&room_id) { - drop(lifecycle_guard); - if redis_room_incremented { - let redis_key = self.room_counter_key(room_id); - self.rollback_distributed_counter(redis_key).await; - } - return Ok(()); - } - - // Step 3: Commit the room move locally after all checks have passed. - // Without Redis, enforce the room limit under the same shard lock as - // the insert so concurrent local joins cannot oversubscribe the room. - { - let mut room_entry = self.room_connections.entry(room_id).or_default(); - if !self.redis_enabled() && room_entry.len() >= self.limits.max_per_room { + let (transition, last_activity) = + if let Some(mut conn) = self.connections.get_mut(connection_id) { + let current_room_id = conn.room_id; + if current_room_id.as_ref() == Some(&room_id) { drop(lifecycle_guard); if redis_room_incremented { let redis_key = self.room_counter_key(room_id); self.rollback_distributed_counter(redis_key).await; } - return Err(format!( - "Room at capacity ({} connections)", - self.limits.max_per_room - )); + return Ok(()); } - room_entry.push(connection_id.to_string()); - } - if let Some(ref old_room) = current_room_id { - if let Some(mut old_room_conns) = self.room_connections.get_mut(old_room) { - old_room_conns.retain(|id| id != connection_id); - if old_room_conns.is_empty() { - drop(old_room_conns); - self.room_connections.remove(old_room); + // Step 3: Commit the room move locally after all checks have passed. + // Without Redis, enforce the room limit under the same shard lock as + // the insert so concurrent local joins cannot oversubscribe the room. + { + let mut room_entry = self.room_connections.entry(room_id).or_default(); + if !self.redis_enabled() && room_entry.len() >= self.limits.max_per_room { + drop(lifecycle_guard); + if redis_room_incremented { + let redis_key = self.room_counter_key(room_id); + self.rollback_distributed_counter(redis_key).await; + } + return Err(format!( + "Room at capacity ({} connections)", + self.limits.max_per_room + )); } + room_entry.push(connection_id.to_string()); } - } - conn.room_id = Some(room_id); - conn.last_activity = Instant::now(); - ( - Some(RoomTransition { - previous_room_id: current_room_id, - room_id, - }), - Some(conn.last_activity), - ) - } else { - drop(lifecycle_guard); - if redis_room_incremented { - let redis_key = self.room_counter_key(room_id); - self.rollback_distributed_counter(redis_key).await; - } - return Err("Connection not found".to_string()); - }; + if let Some(ref old_room) = current_room_id { + if let Some(mut old_room_conns) = self.room_connections.get_mut(old_room) { + old_room_conns.retain(|id| id != connection_id); + if old_room_conns.is_empty() { + drop(old_room_conns); + self.room_connections.remove(old_room); + } + } + } + + conn.room_id = Some(room_id); + conn.last_activity = Instant::now(); + ( + Some(RoomTransition { + previous_room_id: current_room_id, + room_id, + }), + Some(conn.last_activity), + ) + } else { + drop(lifecycle_guard); + if redis_room_incremented { + let redis_key = self.room_counter_key(room_id); + self.rollback_distributed_counter(redis_key).await; + } + return Err("Connection not found".to_string()); + }; drop(lifecycle_guard); if let Some(last_activity) = last_activity { self.schedule_idle_timeout(connection_id, last_activity); diff --git a/synctv-realtime/src/sync/redis_pubsub.rs b/synctv-realtime/src/sync/redis_pubsub.rs index 8c2eb7c5..1a51bbe4 100644 --- a/synctv-realtime/src/sync/redis_pubsub.rs +++ b/synctv-realtime/src/sync/redis_pubsub.rs @@ -722,9 +722,9 @@ impl RedisPubSub { // database outbox, so fail them back to that retry loop instead // of also retaining them in the in-memory Redis retry buffer. if req.expects_ack() { - req.acknowledge_failure( - format!("Failed to publish event to Redis: {e}"), - ); + req.acknowledge_failure(format!( + "Failed to publish event to Redis: {e}" + )); } else if req.event.is_critical() { if critical_retry_buffer.len() >= MAX_CRITICAL_BUFFER { let mut dropped = critical_retry_buffer.remove(0); @@ -788,9 +788,7 @@ impl RedisPubSub { synctv_core::metrics::cluster::REALTIME_EVENTS_DROPPED .with_label_values(&["retry_buffer_full"]) .inc(); - req.acknowledge_failure( - "Redis retry buffer full", - ); + req.acknowledge_failure("Redis retry buffer full"); continue; // Continue draining, don't break - critical events still need to be collected } retry_buffer.push(req); @@ -1183,7 +1181,14 @@ impl RedisPubSub { let total_skipped = 0usize; for stream_key in &streams_to_catchup { - match process_stream_catchup(self, stream_key, &catchup_start_id, &mut *stream_cursors).await { + match process_stream_catchup( + self, + stream_key, + &catchup_start_id, + &mut *stream_cursors, + ) + .await + { Ok(caught_up) => { total_caught_up += caught_up; } @@ -1200,7 +1205,14 @@ impl RedisPubSub { "Failed to catch up on historical events, retrying" ); tokio::time::sleep(Duration::from_millis(500_u64 * retry)).await; - match process_stream_catchup(self, stream_key, &catchup_start_id, &mut *stream_cursors).await { + match process_stream_catchup( + self, + stream_key, + &catchup_start_id, + &mut *stream_cursors, + ) + .await + { Ok(caught_up) => { total_caught_up += caught_up; retry_ok = true; @@ -1772,7 +1784,11 @@ impl RedisPubSub { | RealtimeEvent::RoomOwnerInactive { .. } | RealtimeEvent::UserLeft { .. } ) { - super::events::publish_admin_event(&self.admin_event_tx, event.clone(), "Redis room"); + super::events::publish_admin_event( + &self.admin_event_tx, + event.clone(), + "Redis room", + ); } // Handle terminal room-wide admin events before dropping local room state. diff --git a/synctv/src/app.rs b/synctv/src/app.rs index ec0fb127..8af0d9d6 100644 --- a/synctv/src/app.rs +++ b/synctv/src/app.rs @@ -52,8 +52,7 @@ use crate::realtime_outbox_dispatcher::start_realtime_outbox_dispatcher; use crate::server::{LivestreamState, Services, SyncTvServer}; use crate::shutdown::{ AuditFlushHook, CacheFenceRepairHook, CacheInvalidationStopHook, ProviderInvalidationHook, - RoomSettingsServiceShutdownHook, SettingsListenHook, ShutdownCoordinator, - SimpleShutdownHook, + RoomSettingsServiceShutdownHook, SettingsListenHook, ShutdownCoordinator, SimpleShutdownHook, }; /// Infrastructure: Redis (optional), Database, `NodeID`. diff --git a/synctv/src/cli.rs b/synctv/src/cli.rs index cd6706be..a972e1cb 100644 --- a/synctv/src/cli.rs +++ b/synctv/src/cli.rs @@ -6643,7 +6643,8 @@ fn build_get_playback_cli_output( for (mode, info) in modes { for (index, playback_media) in info.medias.iter().enumerate() { let is_default = mode == &playback.default_mode - && i32::try_from(index).is_ok_and(|index| info.default_media_index == Some(index)); + && i32::try_from(index) + .is_ok_and(|index| info.default_media_index == Some(index)); let absolute_url = absolutize_cli_url(&playback_media.url, api_base_url.as_deref()); let output = PlaybackPullUrlCliOutput { mode: mode.clone(), @@ -6927,10 +6928,7 @@ fn database_migrate_summary(output: &DatabaseMigrateCliOutput) -> String { } /// Shared helper for formatting database CLI outputs. -fn database_cli_output_summary( - header: &str, - output: &impl DatabaseCliOutputFields, -) -> String { +fn database_cli_output_summary(header: &str, output: &impl DatabaseCliOutputFields) -> String { let mut lines = vec![ header.to_string(), format!("Migration status: {}", output.migration_status()),