From a0b6800c0a51ecb55109ee87c251b6341c34207f Mon Sep 17 00:00:00 2001
From: ForgQi <34411314+ForgQi@users.noreply.github.com>
Date: Wed, 22 Oct 2025 02:13:37 +0800
Subject: [PATCH] Refactor async worker status and improve file move logic

Refactored worker status management to use async RwLock and updated all related methods to be async. Improved file move logic in hook steps to support cross-filesystem moves by falling back to copy+delete when necessary. Updated API endpoints, download/upload logic, and monitor state transitions to be fully async and more robust. Also added a Debug implementation for Credentials to avoid logging sensitive data, and made minor UI tweaks to disable certain developer options.
---
 Cargo.lock                                    |   5 +-
 Cargo.toml                                    |   2 +-
 app/ui/StreamerActions/PauseButton.tsx        |   2 +-
 app/ui/plugins/developer.tsx                  |   2 +
 crates/biliup-cli/Cargo.toml                  |   1 +
 crates/biliup-cli/src/server/api/endpoints.rs | 116 +++++++++--------
 crates/biliup-cli/src/server/app.rs           |  25 +++-
 .../biliup-cli/src/server/common/download.rs  |  91 +++++++------
 crates/biliup-cli/src/server/common/upload.rs |   5 +-
 crates/biliup-cli/src/server/common/util.rs   |  38 +++---
 crates/biliup-cli/src/server/core/monitor.rs  | 121 ++++++++++++++----
 .../src/server/infrastructure/context.rs      |   8 +-
 .../src/server/infrastructure/models.rs       |  19 +++
 .../server/infrastructure/models/hook_step.rs |  85 +++++++-----
 .../server/infrastructure/service_register.rs |   2 +-
 .../src/server/infrastructure/users.rs        |  13 +-
 crates/biliup-cli/src/server/router.rs        |   2 +-
 17 files changed, 347 insertions(+), 190 deletions(-)

diff --git a/Cargo.lock b/Cargo.lock
index 7e62547..22b92a0 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -395,7 +395,7 @@ dependencies = [
 
 [[package]]
 name = "biliup-cli"
-version = "1.1.16"
+version = "1.1.17"
 dependencies = [
  "async-channel",
  "async-trait",
@@ -414,6 +414,7 @@ dependencies = [
  "image",
  "indexmap",
  "indicatif",
+ "libc",
  "mime_guess",
  "nom 8.0.0",
  "ormlite",
@@ -3641,7 +3642,7 @@ checksum = "a8f112729512f8e442d81f95a8a7ddf2b7c6b8a1a6f509a95864142b30cab2d3"
 
 [[package]]
 name = "stream-gears"
-version = "1.1.16"
+version = "1.1.17"
 dependencies = [
  "async-channel",
  "async-trait",
diff --git a/Cargo.toml b/Cargo.toml
index 9621f53..f4cbef7 100644
--- a/Cargo.toml
+++ b/Cargo.toml
@@ -7,7 +7,7 @@ members = [
 resolver = "3"
 
 [workspace.package]
-version = "1.1.16"
+version = "1.1.17"
 
 [workspace.dependencies]
 serde = { version = "1.0", features = ["derive"] }
diff --git a/app/ui/StreamerActions/PauseButton.tsx b/app/ui/StreamerActions/PauseButton.tsx
index 3e7de46..e0a20e7 100644
--- a/app/ui/StreamerActions/PauseButton.tsx
+++ b/app/ui/StreamerActions/PauseButton.tsx
@@ -45,7 +45,7 @@ export const PauseButton: React.FC<PauseButtonProps> = ({
             onError?.(error as Error);
         }
     };
-    console.log("????????????", streamer.status)
+
     return (
         <Button onClick={handlePause} icon={streamer.status === 'Pause'? <IconPlay />: <IconPause />} theme="borderless" aria-label="暂停" />
     );
diff --git a/app/ui/plugins/developer.tsx b/app/ui/plugins/developer.tsx
index d750932..848a6f6 100644
--- a/app/ui/plugins/developer.tsx
+++ b/app/ui/plugins/developer.tsx
@@ -52,6 +52,7 @@ const Developer: React.FC = () => {
             formApi.setValue('LOGGING.root.handlers', ['console'])
           }}
           showClear={true}
+          disabled
         >
           <Select.Option value="DEBUG">DEBUG</Select.Option>
           <Select.Option value="INFO">INFO</Select.Option>
@@ -72,6 +73,7 @@ const Developer: React.FC = () => {
             formApi.setValue('LOGGING.loggers.biliup.handlers', ['file'])
           }}
           showClear={true}
+          disabled
         >
           <Select.Option value="DEBUG">DEBUG</Select.Option>
           <Select.Option value="INFO">INFO</Select.Option>
diff --git a/crates/biliup-cli/Cargo.toml b/crates/biliup-cli/Cargo.toml
index 141db5a..96018e6 100644
--- a/crates/biliup-cli/Cargo.toml
+++ b/crates/biliup-cli/Cargo.toml
@@ -57,6 +57,7 @@ tower-sessions-sqlx-store = { version = "0.15.0", features = ["sqlite"] }
 async-channel = { version = "2.5.0" }
 nom = "8"
 tokio-util = "0.7.16"
+libc = "0.2.174"
 [features]
 default = []
 python-bridge = []
diff --git a/crates/biliup-cli/src/server/api/endpoints.rs b/crates/biliup-cli/src/server/api/endpoints.rs
index 9176365..29789ab 100644
--- a/crates/biliup-cli/src/server/api/endpoints.rs
+++ b/crates/biliup-cli/src/server/api/endpoints.rs
@@ -25,7 +25,7 @@ use axum::http::StatusCode;
 use axum::response::{IntoResponse, Response};
 use axum::{Json, debug_handler};
 use biliup::credential::Credential;
-use chrono::Local;
+use chrono::{Local, Utc};
 use clap::ValueEnum;
 use error_stack::{Report, ResultExt, bail};
 use ormlite::{Insert, Model};
@@ -47,29 +47,29 @@ pub async fn get_streamers_endpoint(
     State(workers): State<Arc<RwLock<Vec<Arc<Worker>>>>>,
 ) -> Result<Json<Vec<LiveStreamerResponse>>, Response> {
     let live_streamers = get_all_streamer(&pool).await.map_err(report_to_response)?;
-    Ok(Json(
-        live_streamers
+    let mut results = Vec::new();
+    for x in live_streamers {
+        let option = workers
+            .read()
+            .unwrap()
+            .clone()
             .into_iter()
-            .map(|x| {
-                let option = workers
-                    .read()
-                    .unwrap()
-                    .clone()
-                    .into_iter()
-                    .find(|worker| worker.live_streamer.id == x.id);
-                LiveStreamerResponse {
-                    status: option
-                        .as_ref()
-                        .map(|t| format!("{:?}", *t.downloader_status.read().unwrap()))
-                        .unwrap_or_default(),
-                    inner: x,
-                    upload_status: option
-                        .map(|t| format!("{:?}", *t.uploader_status.read().unwrap()))
-                        .unwrap_or_default(),
-                }
-            })
-            .collect(),
-    ))
+            .find(|worker| worker.live_streamer.id == x.id);
+
+        let status = match option.as_ref() {
+            Some(t) => format!("{:?}", *t.downloader_status.read().await),
+            None => String::new(),
+        };
+
+        results.push(LiveStreamerResponse {
+            status,
+            inner: x,
+            upload_status: option
+                .map(|t| format!("{:?}", *t.uploader_status.read().unwrap()))
+                .unwrap_or_default(),
+        });
+    }
+    Ok(Json(results))
 }
 
 pub async fn post_streamers_endpoint(
@@ -147,7 +147,7 @@ pub async fn delete_streamers_endpoint(
     Ok(Json(live_streamers))
 }
 
-#[axum::debug_handler(state = ServiceRegister)]
+// #[axum::debug_handler(state = ServiceRegister)]
 pub async fn pause_streamers_endpoint(
     State(service_register): State<ServiceRegister>,
     State(pool): State<ConnectionPool>,
@@ -166,33 +166,32 @@ pub async fn pause_streamers_endpoint(
             .ok_or(AppError::Unknown)
             .map_err(report_to_response)?;
         let monitor = manager.ensure_monitor(pool.clone());
-        let worker_status = w.downloader_status.read().unwrap().clone();
-        let d = match worker_status {
+        let worker_status = w.downloader_status.read().await.clone();
+        match worker_status {
             WorkerStatus::Working(d) => {
-                w.change_status(Stage::Download, WorkerStatus::Pause);
-                Some(d.clone())
-            }
-            WorkerStatus::Pending => {
-                monitor.rooms_handle.toggle(w.clone()).await;
-                w.change_status(Stage::Download, WorkerStatus::Pause);
-                None
-            }
-            WorkerStatus::Idle => {
-                monitor.rooms_handle.toggle(w.clone()).await;
-                w.change_status(Stage::Download, WorkerStatus::Pause);
-                None
+                d.stop().await.map_err(report_to_response)?;
+                tokio::time::sleep(tokio::time::Duration::from_secs(1)).await;
+                monitor
+                    .rooms_handle
+                    .toggle(w.clone(), WorkerStatus::Pause)
+                    .await;
+                info!(workers=?&w.live_streamer.url, "successfully pause live streamers");
             }
             WorkerStatus::Pause => {
-                monitor.rooms_handle.toggle(w.clone()).await;
-                w.change_status(Stage::Download, WorkerStatus::Idle);
-                None
+                monitor
+                    .rooms_handle
+                    .toggle(w.clone(), WorkerStatus::Idle)
+                    .await;
+                info!(workers=?&w.live_streamer.url, "successfully start live streamers");
+            }
+            _ => {
+                monitor
+                    .rooms_handle
+                    .toggle(w.clone(), WorkerStatus::Pause)
+                    .await;
+                info!("非预期状态")
             }
         };
-
-        if let Some(d) = d {
-            d.stop().await.map_err(report_to_response)?;
-        }
-        info!(workers=?&w.live_streamer.url, "successfully pause live streamers");
     }
 
     Ok(Json(()))
@@ -513,20 +512,23 @@ pub async fn get_videos() -> Result<Json<Vec<serde_json::Value>>, Response> {
     Ok(Json(file_list))
 }
 
+// #[axum::debug_handler(state = ServiceRegister)]
 pub async fn get_status(
     State(service_register): State<ServiceRegister>,
     State(workers): State<Arc<RwLock<Vec<Arc<Worker>>>>>,
     State(config): State<Arc<RwLock<Config>>>,
     State(actor_handle): State<Arc<ActorHandle>>,
 ) -> Result<Json<serde_json::Value>, Response> {
+    let workers_clone = workers.read().unwrap().clone(); // 如果 Vec 本身可以克隆
+
     let mut sw = Vec::new();
-    for worker in workers.read().unwrap().iter() {
+    for worker in &workers_clone {
         sw.push(serde_json::json!({
-        "downloader_status": format!("{:?}", worker.downloader_status.read().unwrap()),
-        "uploader_status": format!("{:?}", worker.uploader_status.read().unwrap()),
-        "live_streamer": worker.live_streamer,
-        "upload_streamer": worker.upload_streamer,
-        }))
+            "downloader_status": format!("{:?}", worker.downloader_status.read().await),
+            "uploader_status": format!("{:?}", worker.uploader_status.read().unwrap()),
+            "live_streamer": worker.live_streamer,
+            "upload_streamer": worker.upload_streamer,
+        }));
     }
 
     Ok(Json(serde_json::json!({
@@ -573,10 +575,14 @@ pub async fn post_uploads(
     if !videos.is_empty() {
         let recorder = Recorder::new(
             upload_config.title.clone(),
-            &upload_config.template_name,
-            "stream_title",
+            StreamerInfo::new(
+                &upload_config.template_name,
+                "stream_title",
+                "",
+                Utc::now(),
+                "",
+            ),
             "",
-            Local::now(),
         );
         let studio = build_studio(&upload_config, &bilibili, videos, recorder)
             .await
diff --git a/crates/biliup-cli/src/server/app.rs b/crates/biliup-cli/src/server/app.rs
index 7af0328..76950e5 100644
--- a/crates/biliup-cli/src/server/app.rs
+++ b/crates/biliup-cli/src/server/app.rs
@@ -19,7 +19,7 @@ use tower_http::cors::{AllowMethods, CorsLayer};
 use tower_sessions::cookie::Key;
 use tower_sessions::{ExpiredDeletion, Expiry, SessionManagerLayer};
 use tower_sessions_sqlx_store::SqliteStore;
-use tracing::info;
+use tracing::{error, info};
 
 /// 应用程序控制器，负责启动和管理Web服务器
 pub struct ApplicationController;
@@ -48,6 +48,7 @@ impl ApplicationController {
         // 配置会话管理层
         let session_layer = SessionManagerLayer::new(session_store)
             .with_secure(false)
+            .with_name("biliup.sid")
             .with_expiry(Expiry::OnInactivity(Duration::days(7)));
         // .with_signed(key);
 
@@ -93,10 +94,24 @@ impl ApplicationController {
             .attach("error while starting API server")?;
 
         // 等待会话清理任务完成
-        deletion_task
-            .await
-            .change_context(AppError::Unknown)?
-            .change_context(AppError::Unknown)?;
+        match deletion_task.await {
+            Ok(Ok(val)) => { /* 正常完成 */ }
+            Ok(Err(e)) => {
+                // 真正业务错误
+                return Err(e).change_context(AppError::Unknown);
+            }
+            Err(join_err) if join_err.is_cancelled() => {
+                info!("Deletion task cancelled on shutdown");
+            }
+            Err(join_err) if join_err.is_panic() => {
+                error!("Deletion task panicked: {join_err}");
+                return Err(AppError::Unknown.into());
+            }
+            Err(join_err) => {
+                error!("Join error: {join_err}");
+                return Err(AppError::Unknown.into());
+            }
+        }
 
         Ok(())
     }
diff --git a/crates/biliup-cli/src/server/common/download.rs b/crates/biliup-cli/src/server/common/download.rs
index 502102a..967aef6 100644
--- a/crates/biliup-cli/src/server/common/download.rs
+++ b/crates/biliup-cli/src/server/common/download.rs
@@ -46,36 +46,41 @@ impl DownloadGuard {
         rooms_handle: Arc<RoomsHandle>,
         downloader: Arc<dyn Downloader + Send + Sync>,
     ) -> Self {
-        // 更新工作器状态为工作中
-        worker.change_status(Stage::Download, WorkerStatus::Working(downloader.clone()));
         Self {
             worker,
             danmaku_client,
             rooms_handle,
         }
     }
-}
 
-impl Drop for DownloadGuard {
-    fn drop(&mut self) {
+    async fn cleanup(&mut self) {
         // 异步清理任务
         let danmaku = self.danmaku_client.clone();
         let rooms_handle = self.rooms_handle.clone();
         let worker = self.worker.clone();
 
-        tokio::spawn(async move {
-            if let Some(client) = danmaku {
-                if let Err(e) = client.stop().await {
-                    error!("Error stopping danmaku client: {}", e);
-                }
+        if let Some(client) = danmaku {
+            if let Err(e) = client.stop().await {
+                error!("Error stopping danmaku client: {}", e);
             }
-            if let WorkerStatus::Pause = *worker.downloader_status.read().unwrap() {
-            } else {
-                // 确保状态更新和资源清理
-                worker.change_status(Stage::Download, WorkerStatus::Idle);
-                rooms_handle.toggle(worker).await;
-            };
-        });
+        }
+
+        // 确保状态更新和资源清理
+        rooms_handle
+            .toggle(worker.clone(), WorkerStatus::Idle)
+            .await;
+        info!(
+            "{} => {:?}",
+            worker.live_streamer.url,
+            worker.downloader_status.read().await
+        );
+    }
+}
+
+impl Drop for DownloadGuard {
+    fn drop(&mut self) {
+        let worker = self.worker.clone();
+        info!("{}完成资源清理", worker.live_streamer.url);
     }
 }
 
@@ -256,6 +261,7 @@ pub struct DownloadTask {
 struct DownloadComponents {
     downloader: Arc<dyn Downloader + Send + Sync>,
     danmaku_client: Option<Arc<dyn Downloader>>,
+    uploader: Sender<UploaderMessage>,
 }
 
 impl DownloadTask {
@@ -271,20 +277,9 @@ impl DownloadTask {
         }
     }
 
-    pub async fn execute(self, uploader: &Sender<UploaderMessage>) -> AppResult<()> {
-        // 初始化组件
-        let components = self.initialize_components().await?;
-
-        // 创建守卫确保清理
-        let _guard = DownloadGuard::new(
-            self.ctx.worker.clone(),
-            components.danmaku_client.clone(),
-            self.rooms_handle.clone(),
-            components.downloader.clone(),
-        );
-
+    pub async fn execute(self, components: DownloadComponents) -> AppResult<()> {
         // 创建事件处理器
-        let processor = SegmentEventProcessor::new(uploader.clone(), self.ctx.clone());
+        let processor = SegmentEventProcessor::new(components.uploader.clone(), self.ctx.clone());
 
         // 启动弹幕客户端
         if let Some(ref client) = components.danmaku_client {
@@ -310,7 +305,7 @@ impl DownloadTask {
         Ok(())
     }
 
-    async fn initialize_components(&self) -> AppResult<DownloadComponents> {
+    async fn initialize_components(&self, uploader: Sender<UploaderMessage>) -> DownloadComponents {
         // 获取配置和主播信息
         let config = self.ctx.worker.get_config();
         let streamer = self.ctx.worker.get_streamer();
@@ -330,10 +325,11 @@ impl DownloadTask {
             .create_downloader(stream_info, config, self.ctx.recorder.clone())
             .await;
 
-        Ok(DownloadComponents {
+        DownloadComponents {
             downloader,
             danmaku_client,
-        })
+            uploader,
+        }
     }
 
     async fn start_danmaku(&self, client: &Arc<dyn Downloader>) -> AppResult<()> {
@@ -365,9 +361,7 @@ impl DActor {
     /// 运行Actor主循环，处理接收到的消息
     pub(crate) async fn run(&mut self) {
         while let Ok(msg) = self.receiver.recv().await {
-            if let Err(e) = self.handle_message(msg).await {
-                error!("Error handling message: {}", e);
-            }
+            self.handle_message(msg).await
         }
     }
 
@@ -375,16 +369,33 @@ impl DActor {
     ///
     /// # 参数
     /// * `msg` - 要处理的下载消息
-    async fn handle_message(&mut self, msg: DownloaderMessage) -> AppResult<()> {
+    async fn handle_message(&mut self, msg: DownloaderMessage) {
         match msg {
             DownloaderMessage::Start(plugin, ctx, rooms_handle) => {
+                let worker = ctx.worker.clone();
                 // 创建下载任务
-                let task = DownloadTask::new(plugin, ctx, rooms_handle);
+                let task = DownloadTask::new(plugin, ctx, rooms_handle.clone());
+
+                // 初始化组件
+                let components = task.initialize_components(self.sender.clone()).await;
+                // 创建守卫确保清理
+                let mut guard = DownloadGuard::new(
+                    worker.clone(),
+                    components.danmaku_client.clone(),
+                    rooms_handle.clone(),
+                    components.downloader.clone(),
+                );
+
+                // 更新工作器状态为工作中
+                rooms_handle
+                    .toggle(worker, WorkerStatus::Working(components.downloader.clone()))
+                    .await;
 
                 // 执行下载（使用 Result 链式处理）
-                task.execute(&self.sender).await?;
+                let result = task.execute(components).await;
 
-                Ok(())
+                guard.cleanup().await;
+                info!("Handling message: {:?} done", result);
             }
         }
     }
diff --git a/crates/biliup-cli/src/server/common/upload.rs b/crates/biliup-cli/src/server/common/upload.rs
index 2e1be08..5abb139 100644
--- a/crates/biliup-cli/src/server/common/upload.rs
+++ b/crates/biliup-cli/src/server/common/upload.rs
@@ -384,7 +384,8 @@ impl UActor {
         match msg {
             UploaderMessage::SegmentEvent(rx, ctx) => {
                 ctx.worker
-                    .change_status(Stage::Upload, WorkerStatus::Pending);
+                    .change_status(Stage::Upload, WorkerStatus::Pending)
+                    .await;
                 let inspect = rx.inspect(|f| {
                     let pool = ctx.pool.clone();
                     let streamer_info_id = ctx.stream_info.streamer_info.id;
@@ -417,7 +418,7 @@ impl UActor {
                     // 可以添加错误通知机制
                 }
                 info!(url=ctx.stream_info.streamer_info.url, result=?result, "后处理执行完毕：Finished processing segment event");
-                ctx.worker.change_status(Stage::Upload, WorkerStatus::Idle);
+                ctx.worker.change_status(Stage::Upload, WorkerStatus::Idle).await;
             }
         }
     }
diff --git a/crates/biliup-cli/src/server/common/util.rs b/crates/biliup-cli/src/server/common/util.rs
index f07f61c..5e71afd 100644
--- a/crates/biliup-cli/src/server/common/util.rs
+++ b/crates/biliup-cli/src/server/common/util.rs
@@ -1,3 +1,4 @@
+use crate::server::infrastructure::models::StreamerInfo;
 use chrono::{DateTime, Duration, Local};
 use serde::{Deserialize, Serialize};
 use std::path::{Path, PathBuf};
@@ -8,30 +9,18 @@ use url::Url;
 pub struct Recorder {
     ///  filename_prefix   文件名前缀模板
     pub filename_prefix: Option<String>,
-    /// 主播名称
-    pub fname: String,
-    /// 直播间标题
-    pub room_title: String,
+    /// 直播间信息
+    pub streamer_info: StreamerInfo,
     /// 录制后保存文件格式 (mp4, ts, mkv, flv)，不含点
     pub suffix: String,
-    /// 开播时间
-    date_time: DateTime<Local>,
 }
 
 impl Recorder {
-    pub fn new(
-        filename_prefix: Option<String>,
-        fname: &str,
-        room_title: &str,
-        suffix: &str,
-        date_time: DateTime<Local>,
-    ) -> Self {
+    pub fn new(filename_prefix: Option<String>, streamer_info: StreamerInfo, suffix: &str) -> Self {
         Self {
             filename_prefix,
-            fname: fname.to_string(),
-            room_title: room_title.to_string(),
+            streamer_info,
             suffix: suffix.to_string(),
-            date_time,
         }
     }
 
@@ -40,15 +29,16 @@ impl Recorder {
         let raw = if let Some(prefix) = &self.filename_prefix {
             self.template_with(prefix)
         } else {
-            format!("{}%Y-%m-%dT%H_%M_%S", self.fname)
+            format!("{}%Y-%m-%dT%H_%M_%S", self.streamer_info.name)
         };
         sanitize_filename(&raw)
     }
 
     fn template_with(&self, template: &str) -> String {
         template
-            .replace("{streamer}", &self.fname)
-            .replace("{title}", &self.room_title)
+            .replace("{streamer}", &self.streamer_info.name)
+            .replace("{title}", &self.streamer_info.title)
+            .replace("{url}", &self.streamer_info.url)
     }
 
     /// 生成“基名”（不带扩展名），时间冲突时按秒+1继续尝试，直到唯一
@@ -68,11 +58,17 @@ impl Recorder {
     /// 生成“基名”（不带扩展名）
     pub fn format_filename(&self) -> String {
         let template = self.filename_template();
-        self.date_time.format(&template).to_string()
+        self.streamer_info
+            .date
+            .with_timezone(&Local)
+            .format(&template)
+            .to_string()
     }
 
     pub fn format(&self, template: &str) -> String {
-        self.date_time
+        self.streamer_info
+            .date
+            .with_timezone(&Local)
             .format(&self.template_with(template))
             .to_string()
     }
diff --git a/crates/biliup-cli/src/server/core/monitor.rs b/crates/biliup-cli/src/server/core/monitor.rs
index d899136..9c3a3ac 100644
--- a/crates/biliup-cli/src/server/core/monitor.rs
+++ b/crates/biliup-cli/src/server/core/monitor.rs
@@ -13,7 +13,7 @@ use std::sync::Arc;
 use std::time::Duration;
 use tokio::sync::oneshot;
 use tokio::task::JoinHandle;
-use tracing::{error, info};
+use tracing::{debug, error, info};
 
 /// 启动客户端监控循环
 ///
@@ -58,7 +58,8 @@ async fn start_client(
                     };
                     info!(url = url, "room: is live -> 开播了");
                     // 更新状态为等待中
-                    room.change_status(Stage::Download, WorkerStatus::Pending);
+                    room.change_status(Stage::Download, WorkerStatus::Pending)
+                        .await;
                     stream_info.streamer_info = insert;
 
                     let streamer = room.get_streamer();
@@ -71,10 +72,8 @@ async fn start_client(
                         streamer
                             .filename_prefix
                             .or(room.get_config().filename_prefix.clone()),
-                        &streamer.remark,
-                        &stream_info.streamer_info.title,
+                        stream_info.streamer_info.clone(),
                         &suffix,
-                        stream_info.streamer_info.date.with_timezone(&Local),
                     );
                     // 修改 ctx
                     ctx.stream_info = stream_info;
@@ -90,8 +89,7 @@ async fn start_client(
                         .await
                         .is_ok()
                     {
-                        // 切换房间状态
-                        rooms_handle.toggle(room).await;
+                        info!("成功开始录制 {}", url)
                     }
                 }
                 Ok(StreamStatus::Offline) => {}
@@ -229,11 +227,14 @@ impl RoomsHandle {
     ///
     /// # 参数
     /// * `worker` - 要切换的工作器
-    pub async fn toggle(&self, worker: Arc<Worker>) {
-        let msg = ActorMessage::Toggle(worker.clone());
+    pub async fn toggle(&self, worker: Arc<Worker>, status: WorkerStatus) {
+        let (send, recv) = oneshot::channel();
+
+        let msg = ActorMessage::Toggle(send, worker.clone(), status);
 
         // 忽略发送错误
         let _ = self.sender.send(msg).await;
+        recv.await.expect("Actor task has been killed")
     }
 }
 
@@ -252,7 +253,7 @@ enum ActorMessage {
         id: i64,
     },
     /// 切换工作器状态
-    Toggle(Arc<Worker>),
+    Toggle(oneshot::Sender<()>, Arc<Worker>, WorkerStatus),
 }
 
 /// 房间Actor
@@ -282,12 +283,12 @@ impl RoomsActor {
     /// 运行Actor主循环
     async fn run(&mut self) {
         while let Ok(msg) = self.receiver.recv().await {
-            self.handle_message(msg);
+            self.handle_message(msg).await;
         }
     }
 
     /// 处理接收到的消息
-    fn handle_message(&mut self, msg: ActorMessage) {
+    async fn handle_message(&mut self, msg: ActorMessage) {
         match msg {
             ActorMessage::NextRoom { respond_to } => {
                 // `let _ =` 忽略发送时的任何错误
@@ -303,8 +304,10 @@ impl RoomsActor {
                 // 如果使用`select!`宏取消等待响应，可能会发生这种情况
                 let _ = respond_to.send(self.del(id));
             }
-            ActorMessage::Toggle(worker) => {
-                self.toggle_keep_order(&worker);
+            ActorMessage::Toggle(sender, worker, status) => {
+                self.toggle_keep_order(&worker, status).await;
+                // `let _ =` 忽略发送时的任何错误
+                let _ = sender.send(());
             }
         }
     }
@@ -342,19 +345,83 @@ impl RoomsActor {
     }
 
     /// 切换工作器状态，保持顺序
-    fn toggle_keep_order(&mut self, worker: &Arc<Worker>) -> bool {
-        // 从活跃列表移动到等待列表
-        if let Some(i) = self.rooms.iter().position(|x| x == worker) {
-            let val = self.rooms.remove(i); // 保序，但O(n)
-            self.waiting.push(val);
-            true
-        } else if let Some(i) = self.waiting.iter().position(|x| x == worker) {
-            // 从等待列表移动到活跃列表
-            let val = self.waiting.swap_remove(i);
-            self.rooms.push(val);
-            true
-        } else {
-            false
+    async fn toggle_keep_order(&mut self, worker: &Arc<Worker>, status: WorkerStatus) {
+        let mut write_guard = worker.downloader_status.write().await;
+        match (&*write_guard, &status) {
+            (WorkerStatus::Working(_), WorkerStatus::Idle) => {
+                if let Some(i) = self.waiting.iter().position(|x| x == worker) {
+                    // 从等待列表移动到活跃列表
+                    let val = self.waiting.swap_remove(i);
+                    self.rooms.push(val);
+                    *write_guard = status;
+                } else {
+                    error!(
+                        url = worker.live_streamer.url,
+                        "working 状态应该在等待队列中"
+                    )
+                }
+            }
+            (WorkerStatus::Idle | WorkerStatus::Pending, WorkerStatus::Working(_)) => {
+                // 从活跃列表移动到等待列表
+                if let Some(i) = self.rooms.iter().position(|x| x == worker) {
+                    let val = self.rooms.remove(i); // 保序，但O(n)
+                    self.waiting.push(val);
+                    *write_guard = status;
+                } else {
+                    error!(url = worker.live_streamer.url, "idle状态应该在活跃列表中");
+                }
+            }
+            (WorkerStatus::Pause, WorkerStatus::Idle) => {
+                if let Some(i) = self.waiting.iter().position(|x| x == worker) {
+                    // 从等待列表移动到活跃列表
+                    let val = self.waiting.swap_remove(i);
+                    self.rooms.push(val);
+                    *write_guard = status;
+                } else {
+                    error!(
+                        url = worker.live_streamer.url,
+                        "working 状态应该在等待队列中"
+                    )
+                }
+            }
+            (WorkerStatus::Idle, WorkerStatus::Pause) => {
+                // 从活跃列表移动到等待列表
+                if let Some(i) = self.rooms.iter().position(|x| x == worker) {
+                    let val = self.rooms.remove(i); // 保序，但O(n)
+                    self.waiting.push(val);
+                    *write_guard = status;
+                } else {
+                    error!(url = worker.live_streamer.url, "idle状态应该在活跃列表中");
+                }
+            }
+            (WorkerStatus::Working(_), WorkerStatus::Pause) => {
+                // 从活跃列表移动到等待列表
+                if let Some(i) = self.rooms.iter().position(|x| x == worker) {
+                    let val = self.rooms.remove(i); // 保序，但O(n)
+                    self.waiting.push(val);
+                    *write_guard = status;
+                } else {
+                    error!(url = worker.live_streamer.url, "idle状态应该在活跃列表中");
+                }
+            }
+            remaining => {
+                error!("非法的状态转移: {:?}", remaining);
+            }
         }
+        // worker.change_status(Stage::Download, status).await;
+
+        // // 从活跃列表移动到等待列表
+        // if let Some(i) = self.rooms.iter().position(|x| x == worker) {
+        //     let val = self.rooms.remove(i); // 保序，但O(n)
+        //     self.waiting.push(val);
+        //     true
+        // } else if let Some(i) = self.waiting.iter().position(|x| x == worker) {
+        //     // 从等待列表移动到活跃列表
+        //     let val = self.waiting.swap_remove(i);
+        //     self.rooms.push(val);
+        //     true
+        // } else {
+        //     false
+        // }
     }
 }
diff --git a/crates/biliup-cli/src/server/infrastructure/context.rs b/crates/biliup-cli/src/server/infrastructure/context.rs
index 93f06e8..381f147 100644
--- a/crates/biliup-cli/src/server/infrastructure/context.rs
+++ b/crates/biliup-cli/src/server/infrastructure/context.rs
@@ -48,7 +48,7 @@ impl Context {
 #[derive(Debug)]
 pub struct Worker {
     /// 下载器状态
-    pub downloader_status: RwLock<WorkerStatus>,
+    pub downloader_status: tokio::sync::RwLock<WorkerStatus>,
     /// 上传器状态
     pub uploader_status: RwLock<WorkerStatus>,
     /// 直播主播信息
@@ -76,7 +76,7 @@ impl Worker {
         client: StatelessClient,
     ) -> Self {
         Self {
-            downloader_status: RwLock::new(Default::default()),
+            downloader_status: tokio::sync::RwLock::new(Default::default()),
             uploader_status: Default::default(),
             live_streamer,
             upload_streamer,
@@ -108,10 +108,10 @@ impl Worker {
     /// # 参数
     /// * `stage` - 工作阶段（下载或上传）
     /// * `status` - 新的工作状态
-    pub fn change_status(&self, stage: Stage, status: WorkerStatus) {
+    pub async fn change_status(&self, stage: Stage, status: WorkerStatus) {
         match stage {
             Stage::Download => {
-                *self.downloader_status.write().unwrap() = status;
+                *self.downloader_status.write().await = status;
             }
             Stage::Upload => {
                 *self.uploader_status.write().unwrap() = status;
diff --git a/crates/biliup-cli/src/server/infrastructure/models.rs b/crates/biliup-cli/src/server/infrastructure/models.rs
index 3ba438d..d88611c 100644
--- a/crates/biliup-cli/src/server/infrastructure/models.rs
+++ b/crates/biliup-cli/src/server/infrastructure/models.rs
@@ -25,6 +25,25 @@ pub struct StreamerInfo {
     pub live_cover_path: String,
 }
 
+impl StreamerInfo {
+    pub fn new(
+        name: &str,
+        url: &str,
+        title: &str,
+        date: DateTime<Utc>,
+        live_cover_path: &str,
+    ) -> Self {
+        Self {
+            id: -1,
+            name: name.to_string(),
+            url: url.to_string(),
+            title: title.to_string(),
+            date,
+            live_cover_path: live_cover_path.to_string(),
+        }
+    }
+}
+
 /// 文件列表模型
 /// 存储录制文件的信息
 #[derive(Model, Debug, Clone, Serialize, Deserialize)]
diff --git a/crates/biliup-cli/src/server/infrastructure/models/hook_step.rs b/crates/biliup-cli/src/server/infrastructure/models/hook_step.rs
index 81e1139..d685a36 100644
--- a/crates/biliup-cli/src/server/infrastructure/models/hook_step.rs
+++ b/crates/biliup-cli/src/server/infrastructure/models/hook_step.rs
@@ -104,48 +104,75 @@ impl HookStep {
     async fn move_file(&self, video_paths: &[&Path], target_dir: &str) -> AppResult<()> {
         let target_path = Path::new(target_dir);
 
-        // 创建目标目录（如果不存在）
         if !target_path.exists() {
             fs::create_dir_all(target_path)
                 .await
                 .change_context(AppError::Unknown)?;
         }
 
-        // 移动每个视频文件到目标目录
         for video_path in video_paths {
-            let file_name = video_path
-                .file_name()
-                .ok_or(AppError::Custom("Invalid file name".to_string()))?;
-            let destination = target_path.join(file_name);
+            self.move_single_file(video_path, target_path).await?;
 
-            info!(
-                "Moving {} to {}",
-                video_path.display(),
-                destination.display()
-            );
-            fs::rename(video_path, destination)
-                .await
-                .change_context(AppError::Unknown)?;
-
-            let buf = video_path.with_extension("xml");
-            if buf.exists() {
-                let file_name = buf
-                    .file_name()
-                    .ok_or(AppError::Custom("Invalid file name".to_string()))?;
-                let destination = target_path.join(file_name);
-                fs::rename(&buf, &destination)
-                    .await
-                    .change_context(AppError::Unknown)?;
-                info!(
-                    "移动弹幕文件: Moving {} to {}",
-                    buf.display(),
-                    destination.display()
-                );
+            // 移动对应的 XML 文件
+            let xml_path = video_path.with_extension("xml");
+            if xml_path.exists() {
+                info!("移动弹幕文件: {}", xml_path.display());
+                self.move_single_file(&xml_path, target_path).await?;
             }
         }
         Ok(())
     }
 
+    /// 移动单个文件，支持跨文件系统
+    async fn move_single_file(&self, source: &Path, target_dir: &Path) -> AppResult<()> {
+        let file_name = source
+            .file_name()
+            .ok_or(AppError::Custom("Invalid file name".to_string()))?;
+        let destination = target_dir.join(file_name);
+
+        info!("Moving {} to {}", source.display(), destination.display());
+
+        // 先尝试 rename（快速，仅同文件系统）
+        match fs::rename(source, &destination).await {
+            Ok(_) => Ok(()),
+            Err(e) => {
+                // 检查是否是跨文件系统错误
+                if HookStep::is_cross_device_error(&e) {
+                    info!("检测到跨文件系统操作，使用复制+删除模式");
+                    // 复制文件
+                    fs::copy(source, &destination)
+                        .await
+                        .change_context(AppError::Unknown)?;
+
+                    // 删除原文件
+                    fs::remove_file(source)
+                        .await
+                        .change_context(AppError::Unknown)?;
+
+                    Ok(())
+                } else {
+                    Err(e).change_context(AppError::Unknown)?
+                }
+            }
+        }
+    }
+
+    /// 判断是否是跨设备/文件系统错误
+    fn is_cross_device_error(error: &std::io::Error) -> bool {
+        #[cfg(unix)]
+        {
+            error.raw_os_error() == Some(libc::EXDEV)
+        }
+        #[cfg(windows)]
+        {
+            error.raw_os_error() == Some(17) // ERROR_NOT_SAME_DEVICE
+        }
+        #[cfg(not(any(unix, windows)))]
+        {
+            error.kind() == std::io::ErrorKind::Other
+        }
+    }
+
     /// 删除指定的视频文件
     ///
     /// # 参数
diff --git a/crates/biliup-cli/src/server/infrastructure/service_register.rs b/crates/biliup-cli/src/server/infrastructure/service_register.rs
index 150c185..9127192 100644
--- a/crates/biliup-cli/src/server/infrastructure/service_register.rs
+++ b/crates/biliup-cli/src/server/infrastructure/service_register.rs
@@ -136,7 +136,7 @@ impl ServiceRegister {
         if len == 0 {
             *manager.monitor.lock().unwrap() = None;
         }
-        let downloader = match &*removed.downloader_status.read().unwrap() {
+        let downloader = match &*removed.downloader_status.read().await {
             WorkerStatus::Working(downloader) => Some(downloader.clone()),
             _ => None,
         };
diff --git a/crates/biliup-cli/src/server/infrastructure/users.rs b/crates/biliup-cli/src/server/infrastructure/users.rs
index ecc6e9b..7a124b5 100644
--- a/crates/biliup-cli/src/server/infrastructure/users.rs
+++ b/crates/biliup-cli/src/server/infrastructure/users.rs
@@ -46,7 +46,7 @@ impl AuthUser for User {
 
 // 认证凭据结构，用于从表单中提取认证字段
 // 用于与后端进行请求认证
-#[derive(Debug, Clone, Deserialize)]
+#[derive(Clone, Deserialize)]
 pub struct Credentials {
     /// 用户名
     pub username: String,
@@ -56,6 +56,17 @@ pub struct Credentials {
     pub next: Option<String>,
 }
 
+// 手动实现Debug trait以避免意外记录密码哈希
+impl std::fmt::Debug for Credentials {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        f.debug_struct("Credentials")
+            .field("next", &self.next)
+            .field("username", &self.username)
+            .field("password", &"[redacted]")
+            .finish()
+    }
+}
+
 /// 认证后端
 /// 负责处理用户认证相关的数据库操作
 #[derive(Debug, Clone)]
diff --git a/crates/biliup-cli/src/server/router.rs b/crates/biliup-cli/src/server/router.rs
index d37bc39..2081e7e 100644
--- a/crates/biliup-cli/src/server/router.rs
+++ b/crates/biliup-cli/src/server/router.rs
@@ -28,7 +28,7 @@ pub fn router(service_register: ServiceRegister) -> Router<()> {
                 .put(put_streamers_endpoint), // 更新主播
         )
         .route("/v1/streamers/{id}", delete(delete_streamers_endpoint)) // 删除主播
-        .route("/v1/streamers/{id}/pause", put(pause_streamers_endpoint)) // 删除主播
+        .route("/v1/streamers/{id}/pause", put(pause_streamers_endpoint))
         // 配置管理路由
         .route(
             "/v1/configuration",