nyaKazuha/BlueArchiveToolkit
1
0
Fork 0
mirror of https://github.com/Yuyi-Oak/BlueArchiveToolkit.git synced 2026-09-18 06:34:54 +08:00
Code Issues Actions 1 Packages Projects Releases Wiki Activity

fix(api): 补齐 bat-api 控制与后端 RPC

Browse Source
This commit is contained in:
nyaKazuha
2026-07-31 17:01:03 +08:00
parent 20ddd67947
commit 6af7706190
17 changed files with 793 additions and 51 deletions
Download Patch File Download Diff File Expand all files Collapse all files
infrastructure/src/bin/bat_official_sync.rs
+130 -10
View File
@@ -388,6 +388,7 @@ fn run_watch(options: CliOptions) -> anyhow::Result<()> {
registry,
queue: task_tx,
base_config: options.config.clone(),
restart_controller: spawn_daemon_restart_controller,
};
let server =
start_daemon_rpc_server(&daemon_state_dir, Arc::clone(control), context.clone())?;
@@ -757,6 +758,8 @@ struct DaemonRpcAck {
state_dir: PathBuf,
socket_path: PathBuf,
#[serde(skip_serializing_if = "Option::is_none")]
controller_pid: Option<u32>,
#[serde(skip_serializing_if = "Option::is_none")]
force: Option<bool>,
}
@@ -1295,8 +1298,11 @@ struct DaemonTaskContext {
registry: TaskRegistry,
queue: mpsc::Sender<TaskJob>,
base_config: OfficialUpdateConfig,
restart_controller: DaemonRestartController,
}
type DaemonRestartController = fn(&Path) -> anyhow::Result<u32>;
/// 任务 worker：单线程 FIFO 消费任务队列，串行执行官方同步/校验。
///
/// 每个任务执行前获取进程内 `sync_lock`，与 watch 循环互斥（等待而非撞文件锁失败）；
@@ -1393,14 +1399,12 @@ fn canonical_rpc_method(method: &str) -> &str {
fn is_pending_rpc_method(method: &str) -> bool {
// task.create：任务统一由 resource.sync / resource.verify / resource.repair / catalog.refresh
// 等语义方法创建，通用创建接口暂不开放。
// daemon.restart / daemon.clean-stable：CLI 侧按进程生命周期处理；
// live RPC 内不做自重启或在线清理。
// daemon.clean-stable：CLI 侧按进程生命周期处理；
// live RPC 内不做在线清理。
// patch.* / unityfs.*：文件级写入入口已开放；发布级 patch 构建、复杂
// UnityFS 语义编辑和 inspect 等子命令仍未开放。
matches!(
method,
"task.create" | RPC_METHOD_RESTART | RPC_METHOD_CLEAN_STABLE
) || (method.starts_with("patch.") && method != RPC_METHOD_PATCH_APPLY)
matches!(method, "task.create" | RPC_METHOD_CLEAN_STABLE)
|| (method.starts_with("patch.") && method != RPC_METHOD_PATCH_APPLY)
|| (method.starts_with("unityfs.")
&& method != RPC_METHOD_UNITYFS_PATCH_TEXT_ASSET
&& method != RPC_METHOD_UNITYFS_PATCH_STRING_FIELD
@@ -1997,7 +2001,10 @@ fn handle_daemon_rpc_client(
let mut notify_stop_after_response = false;
let response = match serde_json::from_str::<JsonRpcRequest>(&line) {
Ok(request) => {
notify_stop_after_response = request.method == RPC_METHOD_STOP;
notify_stop_after_response = matches!(
canonical_rpc_method(&request.method),
RPC_METHOD_STOP | RPC_METHOD_RESTART
);
handle_daemon_rpc_request(request, &state_dir, &control, &tasks)
}
Err(error) => json_rpc_error(None, -32700, format!("JSON-RPC 请求解析失败：{error}")),
@@ -2081,6 +2088,30 @@ fn dispatch_rpc_method(
rpc_ack_value("stop", "后台停止请求已发送", state_dir, None),
)
}
RPC_METHOD_RESTART => match (tasks.restart_controller)(state_dir) {
Ok(controller_pid) => {
daemon_control_mark_stop_requested(control);
let _ = update_daemon_state_only(state_dir, "restarting");
rpc_envelope_ok(
request_id,
"accepted",
rpc_restart_ack_value(
"restart",
"后台重启控制进程已启动；当前 daemon 会在响应后停止并由 Rust 生命周期入口重启",
state_dir,
controller_pid,
),
)
}
Err(error) => rpc_envelope_error(
request_id,
ApiError::new(
ErrorCode::INTERNAL,
"daemon.restart",
format!("启动后台重启控制进程失败：{error}"),
),
),
},
RPC_METHOD_RELOAD => {
daemon_control_request_reload(control);
rpc_envelope_ok(
@@ -2430,11 +2461,30 @@ fn rpc_ack_value(
message,
state_dir: state_dir.to_path_buf(),
socket_path: daemon_socket_path(state_dir),
controller_pid: None,
force,
})
.unwrap_or(serde_json::Value::Null)
}
fn rpc_restart_ack_value(
command: &'static str,
message: &'static str,
state_dir: &Path,
controller_pid: u32,
) -> serde_json::Value {
serde_json::to_value(DaemonRpcAck {
command,
status: "accepted",
message,
state_dir: state_dir.to_path_buf(),
socket_path: daemon_socket_path(state_dir),
controller_pid: Some(controller_pid),
force: None,
})
.unwrap_or(serde_json::Value::Null)
}
/// 构建 `resource.state` 数据：资源发布根、版本状态、上次同步结果。
/// 读取 daemon 状态文件与资源根目录的版本状态（resource/catalog 只读查询共用）。
fn read_daemon_resource_state(
@@ -2613,6 +2663,15 @@ fn build_catalog_diff_report(state_dir: &Path) -> anyhow::Result<serde_json::Val
.and_then(|state| state.previous_available_version.as_ref());
let previous_snapshot =
previous_record.and_then(|record| read_snapshot(&record.snapshot_path).ok().flatten());
let translation_status_code = if previous_record.is_some() {
ReleaseFlowStatusCode::TranslationQueuedOffline
} else {
ReleaseFlowStatusCode::TranslationUnavailable
};
let (status, status_code, status_phase, status_terminal, status_retryable) =
flow_status_fields(ReleaseFlowStatusCode::ParseCompleted);
let (translation_status, translation_status_code, _, _, _) =
flow_status_fields(translation_status_code);
let current_base = current_snapshot.base_snapshot();
let previous_base = previous_snapshot
@@ -2623,6 +2682,13 @@ fn build_catalog_diff_report(state_dir: &Path) -> anyhow::Result<serde_json::Val
let changed_urls = changed_endpoint_urls(&base_delta);
Ok(serde_json::json!({
"available": true,
"status": status,
"status_code": status_code,
"status_phase": status_phase,
"status_terminal": status_terminal,
"status_retryable": status_retryable,
"translation_status": translation_status,
"translation_status_code": translation_status_code,
"current_version_id": current_record.id,
"previous_version_id": previous_record.map(|record| record.id.clone()),
"previous_snapshot_missing": previous_record.is_some() && previous_snapshot.is_none(),
@@ -3500,6 +3566,29 @@ fn start_daemon_with_options(options: &CliOptions) -> anyhow::Result<DaemonStart
)
}
fn spawn_daemon_restart_controller(state_dir: &Path) -> anyhow::Result<u32> {
validate_runtime_state_dir(state_dir).map_err(anyhow::Error::msg)?;
fs::create_dir_all(state_dir)?;
let executable = env::current_exe()?;
let log = open_append_file(&daemon_log_path(state_dir), PRIVATE_FILE_MODE, "后台日志")
.map_err(anyhow::Error::msg)?;
let log_for_stdout = log.try_clone()?;
let mut command = Command::new(executable);
command
.arg("restart")
.arg("--state-dir")
.arg(state_dir)
.arg("--json")
.arg("--no-progress")
.arg("--no-banner")
.stdin(Stdio::null())
.stdout(Stdio::from(log_for_stdout))
.stderr(Stdio::from(log));
configure_daemon_command(&mut command);
let child = command.spawn()?;
Ok(child.id())
}
fn start_daemon_with_args(
state_dir: PathBuf,
resource_output_root: PathBuf,
@@ -8713,10 +8802,10 @@ mod tests {
assert!(is_pending_rpc_method("patch.build"));
assert!(is_pending_rpc_method("unityfs.inspect"));
assert!(is_pending_rpc_method("task.create"));
assert!(is_pending_rpc_method("daemon.restart"));
assert!(is_pending_rpc_method("daemon.clean-stable"));
// sync/verify/repair、task.cancel/logs、catalog.*、resource.manifest 和
// 文件级 patch/unityfs 写入方法已实现。
// restart、sync/verify/repair、task.cancel/logs、catalog.*、
// resource.manifest 和文件级 patch/unityfs 写入方法已实现。
assert!(!is_pending_rpc_method("daemon.restart"));
assert!(!is_pending_rpc_method("patch.apply"));
assert!(!is_pending_rpc_method("unityfs.patch_text_asset"));
assert!(!is_pending_rpc_method("unityfs.patch_string_field"));
@@ -8738,6 +8827,28 @@ mod tests {
assert!(!is_pending_rpc_method("daemon.doctor"));
}
#[test]
fn dispatch_daemon_restart_starts_controller_and_requests_stop() {
let temp = tempfile::TempDir::new().unwrap();
let control = new_daemon_control();
let envelope = dispatch_rpc_method(
&rpc_request("daemon.restart", None),
temp.path(),
&control,
&test_task_context(),
"req-restart-1".to_string(),
);
let value = serde_json::to_value(&envelope).unwrap();
assert_eq!(value["ok"], true);
assert_eq!(value["status"], "accepted");
assert_eq!(value["data"]["command"], "restart");
assert_eq!(value["data"]["controller_pid"], 4242);
assert_eq!(
wait_for_daemon_wake(Some(&control), Duration::from_millis(1)),
DaemonWake::Stop
);
}
#[test]
fn task_registry_cancel_and_logs() {
let registry = TaskRegistry::new();
@@ -8783,9 +8894,14 @@ mod tests {
registry: TaskRegistry::new(),
queue,
base_config,
restart_controller: test_restart_controller,
}
}
fn test_restart_controller(_state_dir: &Path) -> anyhow::Result<u32> {
Ok(4242)
}
#[test]
fn dispatch_unknown_method_returns_unknown_error_envelope() {
let temp = tempfile::TempDir::new().unwrap();
@@ -8917,6 +9033,7 @@ mod tests {
registry: TaskRegistry::new(),
queue,
base_config,
restart_controller: test_restart_controller,
};
let envelope = dispatch_rpc_method(
@@ -8945,6 +9062,7 @@ mod tests {
registry: TaskRegistry::new(),
queue,
base_config: OfficialUpdateConfig::default(),
restart_controller: test_restart_controller,
};
let envelope = dispatch_rpc_method(
@@ -9004,6 +9122,7 @@ mod tests {
registry: TaskRegistry::new(),
queue,
base_config,
restart_controller: test_restart_controller,
};
let envelope = dispatch_rpc_method(
@@ -10197,6 +10316,7 @@ mod tests {
registry: TaskRegistry::new(),
queue,
base_config: OfficialUpdateConfig::default(),
restart_controller: test_restart_controller,
};
let envelope = dispatch_rpc_method(
&rpc_request("catalog.refresh", None),