chore

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::*;
 

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;

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()))
 }
 

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,
 };
 

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",

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

synctv-api/src/impls/client/file_download.rs

@@ -49,15 +49,18 @@ pub(crate) fn avatar_chunk_stream(
 ) -> impl futures::Stream<Item = Result<synctv_proto::client::UserAvatarObjectResponse, ApiError>>
        + 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<Item = Result<synctv_proto::client::ChatAttachmentObjectResponse, ApiError>>
        + 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<Item = Result<synctv_proto::client::MediaCoverObjectResponse, ApiError>>
        + 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<Item = Result<synctv_proto::client::RoomCoverObjectResponse, ApiError>>
        + 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<Item = Result<synctv_proto::client::PlaylistCoverObjectResponse, ApiError>>
        + 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)]

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(),

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,

synctv-cluster/src/discovery/node_registry.rs

@@ -182,7 +182,6 @@ static REGISTER_REMOTE_NODE_SCRIPT: LazyLock<redis::Script> = 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

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<ProbedNodeIdentity> {
-        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
     }
 }
 

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))
 }

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<i64>, query_description: &str) -> Result<i64> {
@@ -284,8 +284,13 @@ impl ReviewRepository {
         reviewed_by: Option<UserId>,
         reason: &str,
     ) -> Result<u64> {
-        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>(

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;
 

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<u64> {
+pub(super) async fn delete_expired_credentials(pool: &PgPool, buffer_hours: u32) -> Result<u64> {
     if buffer_hours == 0 {
         return Ok(0);
     }

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,

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);

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)

synctv-core/src/service/room/member_admin.rs

@@ -1283,7 +1283,10 @@ impl RoomService {
         params: PermissionWriteParams<'_>,
     ) -> Result<RoomMember> {
         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<PermissionChangedOutboxSnapshot> {
         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?;

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,

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(),

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.

synctv-management/src/mapping.rs

@@ -13,14 +13,12 @@ pub(crate) fn map_user_role(role: i32) -> Result<i32, Status> {
 }
 
 pub(crate) fn map_user_status(status: i32) -> Result<i32, Status> {
-    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<i32, Status> {
-    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)
 }
 

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(

synctv-media-providers/src/alist/client.rs

@@ -213,9 +213,8 @@ impl AlistClient {
                 let response = check_response(response).await?;
                 let resp: AlistResp<T> = 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<T> = 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

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,
 };

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

synctv-proxy/src/slice_cache/backend/file_loader.rs

@@ -175,4 +175,3 @@ async fn load_cache_file(
         }
     }
 }
-

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(())
 }
 

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)

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);

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.

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`.

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()),