diff --git a/docs/api.md b/docs/api.md index a1d7bf9..2a01738 100644 --- a/docs/api.md +++ b/docs/api.md @@ -729,7 +729,9 @@ DELETE /admin/storage/endpoints/{endpoint_id} Authorization: Bearer ``` -凭据加密保存且不通过 API 回传。测试接口执行 Bucket 检查、内部临时对象读写删和公网预签名下载；激活接口会再次测试并原子切换活动端点。活动端点不能直接编辑或删除。 +凭据加密保存且不通过 API 回传。测试接口执行 Bucket 检查、内部临时对象读写删和公网预签名下载；激活接口会再次测试并原子切换活动端点。编辑端点时，后端先测试候选配置并抽查一个历史对象，全部通过后才以并发版本校验方式保存；活动端点编辑成功后保持活动，测试失败则原配置不变。 + +删除接口对活动或仍有关联对象的端点执行软删除：端点立即停止接收新对象并从管理列表隐藏；若删除的是活动端点，新写入自动回退本地。历史对象继续使用保留的加密配置下载和清理，关联对象清空且软删除超过 30 天后，后台任务才彻底移除端点凭据。 列表响应中的 `local_object_count` 与 `local_stored_bytes` 统计当前实际位于应用服务器的成品。存在活动 S3 时，新对象仍优先写入 S3；若本次 S3 写入失败，则自动回退本地并按实际后端完成下载和到期清理。 diff --git a/docs/s3-storage-plan.md b/docs/s3-storage-plan.md index 0377b69..55ff942 100644 --- a/docs/s3-storage-plan.md +++ b/docs/s3-storage-plan.md @@ -94,7 +94,7 @@ Worker 每 5 分钟按 `expires_at` 精确删除对象，删除成功后才删 | Force path style | 开启 | | 签名有效期 | `300` 秒 | -Access Key 和 Secret Key 使用项目现有 AES-256-GCM 机制加密入库，页面只显示掩码。保存后先执行“全链路测试”，再点“验证并启用”。后端激活前会再次执行 `HeadBucket + 内部 PutObject/GetObject + 公网预签名 GET + 内部 DeleteObject`。活动端点或仍有关联对象的端点不能直接编辑、归档，需先新增并启用替代端点，等待旧对象过期后再移除。 +Access Key 和 Secret Key 使用项目现有 AES-256-GCM 机制加密入库，页面只显示掩码。保存后先执行“全链路测试”，再点“验证并启用”。后端激活前会再次执行 `HeadBucket + 内部 PutObject/GetObject + 公网预签名 GET + 内部 DeleteObject`。编辑现有端点时先对候选配置执行同等测试，并抽查历史对象是否仍可访问；测试通过后才保存，活动状态不变。删除采用软删除，立即停止新写入并隐藏端点，但保留历史对象所需的加密配置；历史关联清空 30 天后再彻底移除。若删除活动端点，新文件自动回退应用服务器本地，直到启用其他 S3。 ## 7. 后续部署顺序（本次不执行） diff --git a/frontend/src/pages/admin/AdminStoragePage.vue b/frontend/src/pages/admin/AdminStoragePage.vue index 647ca1c..b1d4e51 100644 --- a/frontend/src/pages/admin/AdminStoragePage.vue +++ b/frontend/src/pages/admin/AdminStoragePage.vue @@ -40,8 +40,16 @@ const emptyForm = () => ({ }) const form = ref(emptyForm()) +const editingEndpoint = computed(() => endpoints.value.find((endpoint) => endpoint.id === editingId.value)) const formTitle = computed(() => (editingId.value ? '编辑存储端点' : '新增存储端点')) -const submitLabel = computed(() => (editingId.value ? '保存并停用待测' : '保存端点')) +const submitLabel = computed(() => (editingId.value ? '测试并保存' : '保存端点')) +const formDescription = computed(() => { + if (!editingId.value) return '新增端点保存后保持停用，请执行全链路测试并手动启用。' + if (editingEndpoint.value?.is_active) { + return '保存前会测试候选配置和一个历史对象；全部通过后替换配置，并保持活动状态。' + } + return '保存前会测试候选配置和一个历史对象；全部通过后替换配置，端点保持停用。' +}) function errorText(err: unknown, fallback: string) { if (err instanceof ApiError) return `[${err.code}] ${err.message}` @@ -121,8 +129,10 @@ async function submitForm() { if (form.value.secret_key.trim()) payload.secret_key = form.value.secret_key.trim() if (editingId.value) { - await updateStorageEndpoint(auth.token, editingId.value, payload) - message.value = '端点已保存并停用，请重新测试后启用。' + const updated = await updateStorageEndpoint(auth.token, editingId.value, payload) + message.value = updated.is_active + ? '配置已保存并通过全链路测试，端点保持活动。' + : '配置已保存并通过全链路测试，端点保持停用。' } else { if (!payload.access_key || !payload.secret_key) { error.value = '新增端点必须填写 Access Key 和 Secret Key。' @@ -183,7 +193,16 @@ async function activate(endpoint: AdminStorageEndpoint) { async function removeEndpoint(endpoint: AdminStorageEndpoint) { if (!auth.token) return - const confirmed = window.confirm(`确定删除“${endpoint.name}”吗？关联对象未清空时后端会拒绝删除。`) + const effects = [`确定删除“${endpoint.name}”吗？删除后端点将立即从管理列表移除。`] + if (endpoint.is_active) { + effects.push('这是当前活动端点；删除后新文件将回退到应用服务器本地，直到启用其他 S3。') + } + if (endpoint.object_count > 0) { + effects.push( + `现有 ${endpoint.object_count} 个历史对象仍可通过原配置下载，关联对象过期清理后凭据才会自动移除。`, + ) + } + const confirmed = window.confirm(effects.join('\n\n')) if (!confirmed) return busy.value = `delete:${endpoint.id}` @@ -305,9 +324,9 @@ onMounted(loadEndpoints)
diff --git a/src/api/admin_storage.rs b/src/api/admin_storage.rs index 02db308..667289e 100644 --- a/src/api/admin_storage.rs +++ b/src/api/admin_storage.rs @@ -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, 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), "存储端点已删除"); + } } diff --git a/src/services/storage.rs b/src/services/storage.rs index 475952a..7fa913a 100644 --- a/src/services/storage.rs +++ b/src/services/storage.rs @@ -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,