237899745/ystp
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: allow storage endpoint updates and deletion

Browse Source
This commit is contained in:
237899745
2026-07-25 16:41:31 +08:00
parent 0fe3d4ce8e
commit d29f43d2e3
5 changed files with 188 additions and 68 deletions
Download Patch File Download Diff File Expand all files Collapse all files

198
src/api/admin_storage.rs
View File

@@ -205,19 +205,6 @@ async fn update_storage_endpoint(
let (_jar, admin_id) = require_admin(&state, jar, &headers, ip).await?;
let existing = storage::get_endpoint(&state, endpoint_id).await?;
ensure_configurable(&existing)?;
if existing.is_active {
return Err(AppError::new(
ErrorCode::InvalidRequest,
"活动端点不能直接编辑，请先新增并启用替代端点",
));
}
let (object_count, _) = endpoint_usage(&state, endpoint_id).await?;
if object_count > 0 {
return Err(AppError::new(
ErrorCode::InvalidRequest,
"端点仍有关联对象，不能修改；请新增替代端点并等待旧对象过期",
));
}
let name = match req.name.as_deref() {
Some(value) => validate_name(value)?,
@@ -240,30 +227,67 @@ async fn update_storage_endpoint(
Some(value) => validate_region(value)?,
None => existing.region.clone(),
};
let access_key_encrypted = match req.access_key.as_deref() {
Some(value) if !value.trim().is_empty() => Some(settings::encrypt_secret(
&state,
&validate_credential(value, "Access Key")?,
)?),
_ => None,
let (access_key_encrypted, access_key_hint_value) = match req.access_key.as_deref() {
Some(value) if !value.trim().is_empty() => {
let value = validate_credential(value, "Access Key")?;
(
settings::encrypt_secret(&state, &value)?,
access_key_hint(&value),
)
}
_ => (
existing.access_key_encrypted.clone(),
existing.access_key_hint.clone(),
),
};
let secret_key_encrypted = match req.secret_key.as_deref() {
Some(value) if !value.trim().is_empty() => Some(settings::encrypt_secret(
&state,
&validate_credential(value, "Secret Key")?,
)?),
_ => None,
Some(value) if !value.trim().is_empty() => {
settings::encrypt_secret(&state, &validate_credential(value, "Secret Key")?)?
}
_ => existing.secret_key_encrypted.clone(),
};
let access_key_hint_value = req
.access_key
.as_deref()
.filter(|value| !value.trim().is_empty())
.map(access_key_hint);
let force_path_style = req.force_path_style.unwrap_or(existing.force_path_style);
let presign_ttl_seconds = validate_presign_ttl(
req.presign_ttl_seconds
.unwrap_or(existing.presign_ttl_seconds),
)?;
let mut candidate = existing.clone();
candidate.name = name.clone();
candidate.internal_endpoint = internal_endpoint.clone();
candidate.public_endpoint = public_endpoint.clone();
candidate.bucket = bucket.clone();
candidate.region = region.clone();
candidate.access_key_encrypted = access_key_encrypted.clone();
candidate.secret_key_encrypted = secret_key_encrypted.clone();
candidate.access_key_hint = access_key_hint_value.clone();
candidate.force_path_style = force_path_style;
candidate.presign_ttl_seconds = presign_ttl_seconds;
let test_result = match storage::test_endpoint(&state, &candidate).await {
Ok(()) => match endpoint_sample_key(&state, endpoint_id).await? {
Some(key) => storage::test_existing_object(&state, &candidate, &key).await,
None => Ok(()),
},
Err(err) => Err(err),
};
if let Err(err) = test_result {
let detail = storage_test_message(&err);
audit_storage_action(
&state,
admin_id,
ip,
"storage_endpoint.update_test_failed",
endpoint_id,
serde_json::json!({ "name": name, "message": detail }),
)
.await;
return Err(AppError::new(
ErrorCode::StorageUnavailable,
format!("新配置测试失败，原配置未修改：{detail}"),
));
}
let endpoint = sqlx::query_as::<_, storage::StorageEndpoint>(
r#"
UPDATE storage_endpoints
@@ -272,18 +296,17 @@ async fn update_storage_endpoint(
public_endpoint = $4,
bucket = $5,
region = $6,
access_key_encrypted = COALESCE($7, access_key_encrypted),
secret_key_encrypted = COALESCE($8, secret_key_encrypted),
access_key_hint = COALESCE($9, access_key_hint),
access_key_encrypted = $7,
secret_key_encrypted = $8,
access_key_hint = $9,
force_path_style = $10,
presign_ttl_seconds = $11,
is_active = false,
last_test_at = NULL,
last_test_ok = NULL,
last_test_at = NOW(),
last_test_ok = true,
last_test_error = NULL,
updated_at = NOW(),
updated_by = $12
WHERE id = $1
WHERE id = $1 AND updated_at = $13 AND deleted_at IS NULL
RETURNING id, name, internal_endpoint, public_endpoint, bucket, region,
access_key_encrypted, secret_key_encrypted, access_key_hint,
force_path_style, presign_ttl_seconds, is_active,
@@ -300,12 +323,19 @@ async fn update_storage_endpoint(
.bind(access_key_encrypted)
.bind(secret_key_encrypted)
.bind(access_key_hint_value)
.bind(req.force_path_style.unwrap_or(existing.force_path_style))
.bind(force_path_style)
.bind(presign_ttl_seconds)
.bind(admin_id)
.fetch_one(&state.db)
.bind(existing.updated_at)
.fetch_optional(&state.db)
.await
.map_err(|err| AppError::new(ErrorCode::Internal, "更新存储端点失败").with_source(err))?;
.map_err(|err| AppError::new(ErrorCode::Internal, "更新存储端点失败").with_source(err))?
.ok_or_else(|| {
AppError::new(
ErrorCode::InvalidRequest,
"端点配置在测试期间发生变化，请重新编辑",
)
})?;
audit_storage_action(
&state,
@@ -313,7 +343,7 @@ async fn update_storage_endpoint(
ip,
"storage_endpoint.update",
endpoint.id,
serde_json::json!({ "name": endpoint.name, "requires_reactivation": true }),
serde_json::json!({ "name": endpoint.name, "kept_active": endpoint.is_active }),
)
.await;
@@ -458,42 +488,42 @@ async fn delete_storage_endpoint(
let (_jar, admin_id) = require_admin(&state, jar, &headers, ip).await?;
let endpoint = storage::get_endpoint(&state, endpoint_id).await?;
ensure_configurable(&endpoint)?;
if endpoint.is_active {
return Err(AppError::new(
ErrorCode::InvalidRequest,
"活动端点不能删除，请先启用其他端点",
));
}
let (object_count, _): (i64, i64) = endpoint_usage(&state, endpoint_id).await?;
if object_count > 0 {
return Err(AppError::new(
ErrorCode::InvalidRequest,
format!("端点仍关联 {object_count} 个对象，需等待对象过期清理后再删除"),
));
}
sqlx::query(
"UPDATE storage_endpoints SET is_active = false, deleted_at = NOW(), updated_at = NOW(), updated_by = $2 WHERE id = $1",
let deleted = sqlx::query(
"UPDATE storage_endpoints SET is_active = false, deleted_at = NOW(), updated_at = NOW(), updated_by = $2 WHERE id = $1 AND updated_at = $3 AND deleted_at IS NULL",
)
.bind(endpoint_id)
.bind(admin_id)
.bind(endpoint.updated_at)
.execute(&state.db)
.await
.map_err(|err| AppError::new(ErrorCode::Internal, "归档存储端点失败").with_source(err))?;
if deleted.rows_affected() == 0 {
return Err(AppError::new(
ErrorCode::InvalidRequest,
"端点配置已发生变化，请刷新后重试",
));
}
audit_storage_action(
&state,
admin_id,
ip,
"storage_endpoint.delete",
endpoint_id,
serde_json::json!({ "name": endpoint.name }),
serde_json::json!({
"name": endpoint.name,
"object_count": object_count,
"was_active": endpoint.is_active,
}),
)
.await;
Ok(Json(Envelope {
success: true,
data: serde_json::json!({ "message": "存储端点已删除" }),
data: serde_json::json!({
"message": delete_endpoint_message(endpoint.is_active, object_count),
}),
}))
}
@@ -541,6 +571,31 @@ async fn endpoint_usage(state: &AppState, endpoint_id: Uuid) -> Result<(i64, i64
.map_err(|err| AppError::new(ErrorCode::Internal, "统计存储使用量失败").with_source(err))
}
async fn endpoint_sample_key(
state: &AppState,
endpoint_id: Uuid,
) -> Result<Option<String>, AppError> {
sqlx::query_scalar(
r#"
SELECT object_key
FROM (
SELECT storage_key AS object_key
FROM task_files
WHERE storage_endpoint_id = $1 AND storage_key IS NOT NULL
UNION ALL
SELECT zip_storage_key AS object_key
FROM tasks
WHERE zip_storage_endpoint_id = $1 AND zip_storage_key IS NOT NULL
) stored_objects
LIMIT 1
"#,
)
.bind(endpoint_id)
.fetch_optional(&state.db)
.await
.map_err(|err| AppError::new(ErrorCode::Internal, "查询存储历史对象失败").with_source(err))
}
async fn local_usage(state: &AppState) -> Result<(i64, i64), AppError> {
sqlx::query_as::<_, (i64, i64)>(
r#"
@@ -724,6 +779,23 @@ fn access_key_hint(value: &str) -> String {
format!("{start}...{end}")
}
fn delete_endpoint_message(was_active: bool, object_count: i64) -> String {
if object_count > 0 {
let fallback = if was_active {
"新文件将回退到本地存储；"
} else {
""
};
return format!(
"存储端点已删除并停止接收新对象；{fallback}仍关联的 {object_count} 个历史对象可继续下载，关联对象清理完成后凭据将自动移除"
);
}
if was_active {
return "存储端点已删除；当前没有活动 S3，新文件将回退到本地存储".to_string();
}
"存储端点已删除".to_string()
}
fn storage_test_message(error: &AppError) -> String {
let Some(source) = error.source.as_deref() else {
return error.message.clone();
@@ -785,4 +857,16 @@ mod tests {
assert_eq!(access_key_hint("ABCD12345678WXYZ"), "ABCD...WXYZ");
assert_eq!(access_key_hint("short"), "********");
}
#[test]
fn deleting_active_endpoint_explains_local_fallback() {
let message = delete_endpoint_message(true, 3);
assert!(message.contains("新文件将回退到本地存储"));
assert!(message.contains("3 个历史对象可继续下载"));
}
#[test]
fn deleting_inactive_unused_endpoint_has_short_message() {
assert_eq!(delete_endpoint_message(false, 0), "存储端点已删除");
}
}

16
src/services/storage.rs
View File

@@ -522,6 +522,22 @@ pub async fn test_endpoint(state: &AppState, endpoint: &StorageEndpoint) -> Resu
Ok(())
}
pub async fn test_existing_object(
state: &AppState,
endpoint: &StorageEndpoint,
key: &str,
) -> Result<(), AppError> {
let client = client_for(state, endpoint, EndpointKind::Internal)?;
client
.head_object()
.bucket(&endpoint.bucket)
.key(key)
.send()
.await
.map_err(|err| storage_error("S3 历史对象访问测试失败", err))?;
Ok(())
}
fn client_for(
state: &AppState,
endpoint: &StorageEndpoint,