feat(admin): manual request-records cleanup with typed confirmation

2026-09-02 01:10:23 +08:00 · 2026-05-13 00:36:43 +08:00
parent 8714d93d4b
commit 15800d7a80
21 changed files with 1064 additions and 28 deletions
--- a/apps/aether-gateway/src/control/route/admin/system_families.rs
+++ b/apps/aether-gateway/src/control/route/admin/system_families.rs
@@ -103,6 +103,26 @@ pub(super) fn classify_admin_system_family_route(
            "admin:system",
            false,
        ))
+    } else if method == http::Method::POST
+        && normalized_path == "/api/admin/system/cleanup/usage/manual"
+    {
+        Some(classified(
+            "admin_proxy",
+            "system_manage",
+            "cleanup_usage_manual",
+            "admin:system",
+            false,
+        ))
+    } else if method == http::Method::GET
+        && normalized_path == "/api/admin/system/cleanup/usage/preview"
+    {
+        Some(classified(
+            "admin_proxy",
+            "system_manage",
+            "cleanup_usage_preview",
+            "admin:system",
+            false,
+        ))
    } else if method == http::Method::POST && normalized_path == "/api/admin/system/purge/config" {
        Some(classified(
            "admin_proxy",
--- a/apps/aether-gateway/src/data/state/runtime.rs
+++ b/apps/aether-gateway/src/data/state/runtime.rs
@@ -978,6 +978,19 @@ impl GatewayDataState {
        }
    }

+    pub(crate) async fn preview_usage_cleanup(
+        &self,
+        window: &UsageCleanupWindow,
+    ) -> Result<aether_data_contracts::repository::usage::UsageCleanupPreviewCounts, DataLayerError>
+    {
+        match &self.usage_writer {
+            Some(repository) => repository.preview_usage_cleanup(window).await,
+            None => {
+                Ok(aether_data_contracts::repository::usage::UsageCleanupPreviewCounts::default())
+            }
+        }
+    }
+
    pub(crate) async fn find_request_usage_by_request_id(
        &self,
        request_id: &str,
--- a/apps/aether-gateway/src/handlers/admin/system/core/system_routes.rs
+++ b/apps/aether-gateway/src/handlers/admin/system/core/system_routes.rs
@@ -204,6 +204,24 @@ pub(super) async fn maybe_build_local_admin_core_system_response(
        return Ok(Some(Json(json!({ "items": records })).into_response()));
    }

+    if decision.route_kind.as_deref() == Some("cleanup_usage_manual")
+        && request_method == http::Method::POST
+        && request_path == "/api/admin/system/cleanup/usage/manual"
+    {
+        return Ok(Some(
+            build_manual_usage_cleanup_response(state, request_body).await?,
+        ));
+    }
+
+    if decision.route_kind.as_deref() == Some("cleanup_usage_preview")
+        && request_method == http::Method::GET
+        && request_path == "/api/admin/system/cleanup/usage/preview"
+    {
+        return Ok(Some(
+            build_manual_usage_cleanup_preview_response(state, request_context).await?,
+        ));
+    }
+
    if let Some((task_kind, action, object_type, object_id)) =
        admin_system_purge_task_for_route_kind(decision.route_kind.as_deref())
    {
@@ -550,3 +568,171 @@ async fn build_admin_system_cleanup_payload(
        "cleaned": cleaned,
    }))
 }
+
+async fn build_manual_usage_cleanup_response(
+    state: &AdminAppState<'_>,
+    request_body: Option<&Bytes>,
+) -> Result<Response<Body>, GatewayError> {
+    let older_than_days = match parse_manual_usage_cleanup_request(request_body) {
+        Ok(value) => value,
+        Err(response) => return Ok(response),
+    };
+
+    match crate::maintenance::run_manual_usage_cleanup_once(
+        &state.app().data,
+        older_than_days,
+        None,
+    )
+    .await
+    {
+        Ok(summary) => {
+            let total = summary
+                .body_externalized
+                .saturating_add(summary.legacy_body_refs_migrated)
+                .saturating_add(summary.body_cleaned)
+                .saturating_add(summary.header_cleaned)
+                .saturating_add(summary.keys_cleaned)
+                .saturating_add(summary.records_deleted);
+            let message = match older_than_days {
+                Some(days) => {
+                    format!("请求记录手动清理完成，清理 {days} 天前的记录，影响 {total} 项")
+                }
+                None => format!("请求记录手动清理完成（按当前策略），影响 {total} 项"),
+            };
+            let payload = json!({
+                "message": message,
+                "requested_older_than_days": older_than_days,
+                "summary": {
+                    "body_externalized": summary.body_externalized,
+                    "legacy_body_refs_migrated": summary.legacy_body_refs_migrated,
+                    "body_cleaned": summary.body_cleaned,
+                    "header_cleaned": summary.header_cleaned,
+                    "keys_cleaned": summary.keys_cleaned,
+                    "records_deleted": summary.records_deleted,
+                },
+                "total_affected": total,
+            });
+            Ok(attach_admin_audit_response(
+                Json(payload).into_response(),
+                "admin_system_usage_cleanup_completed",
+                "manual_usage_cleanup",
+                "usage_cleanup",
+                "global",
+            ))
+        }
+        Err(crate::maintenance::ManualUsageCleanupError::AlreadyRunning) => Ok((
+            http::StatusCode::CONFLICT,
+            Json(json!({
+                "detail": "usage_cleanup_already_running",
+                "message": "已有一次清理正在进行中，请稍后再试",
+            })),
+        )
+            .into_response()),
+        Err(crate::maintenance::ManualUsageCleanupError::DataLayer(err)) => {
+            Err(GatewayError::Internal(err.to_string()))
+        }
+    }
+}
+
+async fn build_manual_usage_cleanup_preview_response(
+    state: &AdminAppState<'_>,
+    request_context: &AdminRequestContext<'_>,
+) -> Result<Response<Body>, GatewayError> {
+    let older_than_days = match parse_older_than_days_query(request_context.query_string()) {
+        Ok(value) => value,
+        Err(response) => return Ok(response),
+    };
+    let preview =
+        crate::maintenance::preview_manual_usage_cleanup(&state.app().data, older_than_days)
+            .await
+            .map_err(|err| GatewayError::Internal(err.to_string()))?;
+    Ok(Json(json!({
+        "requested_older_than_days": preview.requested_older_than_days,
+        "effective_cutoffs": {
+            "detail": preview.detail_cutoff,
+            "compressed": preview.compressed_cutoff,
+            "header": preview.header_cutoff,
+            "log": preview.log_cutoff,
+        },
+        "counts": {
+            "detail": preview.detail_count,
+            "compressed": preview.compressed_count,
+            "header": preview.header_count,
+            "log": preview.log_count,
+        },
+    }))
+    .into_response())
+}
+
+fn parse_manual_usage_cleanup_request(
+    request_body: Option<&Bytes>,
+) -> Result<Option<u32>, Response<Body>> {
+    let Some(body) = request_body else {
+        return Ok(None);
+    };
+    if body.is_empty() {
+        return Ok(None);
+    }
+    let parsed: serde_json::Value = match serde_json::from_slice(body) {
+        Ok(value) => value,
+        Err(err) => {
+            return Err((
+                http::StatusCode::BAD_REQUEST,
+                Json(json!({ "detail": format!("请求体无效 JSON: {err}") })),
+            )
+                .into_response())
+        }
+    };
+    let Some(object) = parsed.as_object() else {
+        return Ok(None);
+    };
+    match object.get("older_than_days") {
+        None | Some(serde_json::Value::Null) => Ok(None),
+        Some(value) => value
+            .as_u64()
+            .and_then(|value| u32::try_from(value).ok())
+            .filter(|days| *days >= 1)
+            .map(Some)
+            .ok_or_else(|| {
+                (
+                    http::StatusCode::BAD_REQUEST,
+                    Json(json!({
+                        "detail": "older_than_days 必须为正整数",
+                    })),
+                )
+                    .into_response()
+            }),
+    }
+}
+
+fn parse_older_than_days_query(query_string: Option<&str>) -> Result<Option<u32>, Response<Body>> {
+    let Some(query) = query_string.filter(|value| !value.is_empty()) else {
+        return Ok(None);
+    };
+    let value = query
+        .split('&')
+        .filter_map(|pair| pair.split_once('='))
+        .find_map(|(key, value)| {
+            if key == "older_than_days" && !value.is_empty() {
+                Some(value)
+            } else {
+                None
+            }
+        });
+    let Some(raw) = value else {
+        return Ok(None);
+    };
+    raw.parse::<u32>()
+        .ok()
+        .filter(|days| *days >= 1)
+        .map(Some)
+        .ok_or_else(|| {
+            (
+                http::StatusCode::BAD_REQUEST,
+                Json(json!({
+                    "detail": "older_than_days 必须为正整数",
+                })),
+            )
+                .into_response()
+        })
+}
--- a/apps/aether-gateway/src/maintenance/mod.rs
+++ b/apps/aether-gateway/src/maintenance/mod.rs
@@ -6,22 +6,24 @@ pub(crate) use runtime::{
    cancel_proxy_upgrade_rollout, clear_proxy_upgrade_rollout_conflicts,
    ensure_provider_key_pool_scores_for_keys, inspect_proxy_upgrade_rollout,
    list_admin_cleanup_run_records, perform_oauth_token_refresh_once,
-    perform_pool_quota_probe_once, perform_provider_checkin_once, rebuild_admin_stats_once,
-    record_completed_cleanup_run, record_proxy_upgrade_traffic_success,
+    perform_pool_quota_probe_once, perform_provider_checkin_once, preview_manual_usage_cleanup,
+    rebuild_admin_stats_once, record_completed_cleanup_run, record_proxy_upgrade_traffic_success,
    restore_proxy_upgrade_rollout_skipped_nodes, retry_proxy_upgrade_rollout_node,
-    run_admin_system_cleanup_once, skip_proxy_upgrade_rollout_node, spawn_audit_cleanup_worker,
-    spawn_db_maintenance_worker, spawn_gemini_file_mapping_cleanup_worker,
-    spawn_oauth_token_refresh_worker, spawn_pending_cleanup_worker, spawn_pool_monitor_worker,
-    spawn_pool_quota_probe_worker, spawn_pool_score_rebuild_worker, spawn_provider_checkin_worker,
+    run_admin_system_cleanup_once, run_manual_usage_cleanup_once, skip_proxy_upgrade_rollout_node,
+    spawn_audit_cleanup_worker, spawn_db_maintenance_worker,
+    spawn_gemini_file_mapping_cleanup_worker, spawn_oauth_token_refresh_worker,
+    spawn_pending_cleanup_worker, spawn_pool_monitor_worker, spawn_pool_quota_probe_worker,
+    spawn_pool_score_rebuild_worker, spawn_provider_checkin_worker,
    spawn_proxy_node_metrics_cleanup_worker, spawn_proxy_node_stale_cleanup_worker,
    spawn_proxy_upgrade_rollout_worker, spawn_request_candidate_cleanup_worker,
    spawn_stats_aggregation_worker, spawn_stats_hourly_aggregation_worker,
    spawn_usage_cleanup_worker, spawn_wallet_daily_usage_aggregation_worker,
    start_admin_request_body_cleanup_task, start_admin_system_purge_task,
    start_proxy_upgrade_rollout, AdminCleanupRunRecord, AdminCleanupTaskKind,
-    AdminStatsRebuildSummary, AdminSystemCleanupSummary, OAuthTokenRefreshRunSummary,
-    PoolQuotaProbeRunSummary, ProviderCheckinRunSummary, ProxyUpgradeRolloutCancelSummary,
-    ProxyUpgradeRolloutConflictClearSummary, ProxyUpgradeRolloutNodeActionSummary,
-    ProxyUpgradeRolloutProbeConfig, ProxyUpgradeRolloutSkippedRestoreSummary,
-    ProxyUpgradeRolloutStatus, ProxyUpgradeRolloutTrackedNodeState,
+    AdminStatsRebuildSummary, AdminSystemCleanupSummary, ManualUsageCleanupError,
+    OAuthTokenRefreshRunSummary, PoolQuotaProbeRunSummary, ProviderCheckinRunSummary,
+    ProxyUpgradeRolloutCancelSummary, ProxyUpgradeRolloutConflictClearSummary,
+    ProxyUpgradeRolloutNodeActionSummary, ProxyUpgradeRolloutProbeConfig,
+    ProxyUpgradeRolloutSkippedRestoreSummary, ProxyUpgradeRolloutStatus,
+    ProxyUpgradeRolloutTrackedNodeState,
 };
--- a/apps/aether-gateway/src/maintenance/runtime.rs
+++ b/apps/aether-gateway/src/maintenance/runtime.rs
@@ -56,7 +56,7 @@ use audit_cleanup::*;
 pub(crate) use cleanup_runs::{
    list_admin_cleanup_run_records, record_completed_cleanup_run, record_failed_cleanup_run,
    start_admin_request_body_cleanup_task, start_admin_system_purge_task, AdminCleanupRunRecord,
-    AdminCleanupTaskKind,
+    AdminCleanupTaskKind, USAGE_CLEANUP_KIND,
 };
 use config::*;
 use db_maintenance::*;
@@ -90,9 +90,11 @@ pub(crate) use proxy_upgrade_rollout::{
 };
 use request_candidate_cleanup::*;
 use runners::*;
+pub(crate) use runners::{run_manual_usage_cleanup_once, ManualUsageCleanupError};
 use schedule::*;
 use stats_daily::*;
 use stats_hourly::*;
+pub(crate) use usage_cleanup::preview_manual_usage_cleanup;
 use usage_cleanup::*;
 use wallet_daily_usage::*;
 pub(crate) use workers::*;
--- a/apps/aether-gateway/src/maintenance/runtime/cleanup_runs.rs
+++ b/apps/aether-gateway/src/maintenance/runtime/cleanup_runs.rs
@@ -15,6 +15,7 @@ const CLEANUP_RUN_HISTORY_KEY: &str = "admin_cleanup_run_history";
 const CLEANUP_RUN_HISTORY_LIMIT: usize = 50;
 const REQUEST_BODY_PROGRESS_UPDATE_BATCHES: usize = 10;

+pub(crate) const USAGE_CLEANUP_KIND: &str = "usage_cleanup";
 #[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
 pub(crate) struct AdminCleanupRunRecord {
    pub(crate) id: String,
--- a/apps/aether-gateway/src/maintenance/runtime/config.rs
+++ b/apps/aether-gateway/src/maintenance/runtime/config.rs
@@ -111,12 +111,33 @@ pub(super) async fn usage_cleanup_settings(
 pub(super) fn usage_cleanup_window(
    now_utc: DateTime<Utc>,
    settings: UsageCleanupSettings,
+) -> UsageCleanupWindow {
+    usage_cleanup_window_with_override(now_utc, settings, None)
+}
+
+/// Clamp is non-aggressive: each tier's cutoff becomes `max(policy_cutoff, now - override)`.
+/// A later cutoff = fewer records deleted, so the override can only make cleanup more
+/// conservative than the configured retention, never more destructive.
+pub(super) fn usage_cleanup_window_with_override(
+    now_utc: DateTime<Utc>,
+    settings: UsageCleanupSettings,
+    override_older_than: Option<chrono::Duration>,
 ) -> UsageCleanupWindow {
    let minutes = |days: u64| chrono::Duration::days(i64::try_from(days).unwrap_or(i64::MAX));
-    UsageCleanupWindow {
+    let policy = UsageCleanupWindow {
        detail_cutoff: now_utc - minutes(settings.detail_retention_days),
        compressed_cutoff: now_utc - minutes(settings.compressed_retention_days),
        header_cutoff: now_utc - minutes(settings.header_retention_days),
        log_cutoff: now_utc - minutes(settings.log_retention_days),
+    };
+    let Some(override_duration) = override_older_than else {
+        return policy;
+    };
+    let manual_cutoff = now_utc - override_duration;
+    UsageCleanupWindow {
+        detail_cutoff: policy.detail_cutoff.max(manual_cutoff),
+        compressed_cutoff: policy.compressed_cutoff.max(manual_cutoff),
+        header_cutoff: policy.header_cutoff.max(manual_cutoff),
+        log_cutoff: policy.log_cutoff.max(manual_cutoff),
    }
 }
--- a/apps/aether-gateway/src/maintenance/runtime/runners.rs
+++ b/apps/aether-gateway/src/maintenance/runtime/runners.rs
@@ -13,8 +13,9 @@ use super::{
    cleanup_stale_proxy_nodes_once, collect_proxy_upgrade_rollout_probes, now_unix_secs,
    perform_db_maintenance_once, perform_provider_checkin_once, perform_stats_aggregation_once,
    perform_stats_hourly_aggregation_once, perform_usage_cleanup_once,
-    perform_wallet_daily_usage_aggregation_once, record_completed_cleanup_run,
-    record_failed_cleanup_run, record_proxy_upgrade_traffic_success, summarize_database_pool,
+    perform_usage_cleanup_once_with_override, perform_wallet_daily_usage_aggregation_once,
+    record_completed_cleanup_run, record_failed_cleanup_run, record_proxy_upgrade_traffic_success,
+    summarize_database_pool,
 };

 pub(super) async fn run_audit_cleanup_once(data: &GatewayDataState) -> Result<(), DataLayerError> {
@@ -298,6 +299,110 @@ pub(super) async fn run_usage_cleanup_once(data: &GatewayDataState) -> Result<()
    Ok(())
 }

+pub(crate) async fn run_manual_usage_cleanup_once(
+    data: &GatewayDataState,
+    override_older_than_days: Option<u32>,
+    actor_user_id: Option<String>,
+) -> Result<aether_data_contracts::repository::usage::UsageCleanupSummary, ManualUsageCleanupError>
+{
+    use super::{list_admin_cleanup_run_records, USAGE_CLEANUP_KIND};
+
+    let existing = list_admin_cleanup_run_records(data)
+        .await
+        .map_err(ManualUsageCleanupError::DataLayer)?;
+    if existing
+        .iter()
+        .any(|record| record.kind == USAGE_CLEANUP_KIND && record.status == "processing")
+    {
+        return Err(ManualUsageCleanupError::AlreadyRunning);
+    }
+
+    let started_at_unix_secs = now_unix_secs();
+    let started_at = Instant::now();
+    let override_duration =
+        override_older_than_days.map(|days| chrono::Duration::days(i64::from(days)));
+    let summary = match perform_usage_cleanup_once_with_override(data, override_duration).await {
+        Ok(summary) => summary,
+        Err(err) => {
+            record_failed_cleanup_run(
+                data,
+                USAGE_CLEANUP_KIND,
+                "manual",
+                started_at_unix_secs,
+                started_at,
+                &err,
+            )
+            .await;
+            return Err(ManualUsageCleanupError::DataLayer(err));
+        }
+    };
+    let total = summary
+        .body_externalized
+        .saturating_add(summary.legacy_body_refs_migrated)
+        .saturating_add(summary.body_cleaned)
+        .saturating_add(summary.header_cleaned)
+        .saturating_add(summary.keys_cleaned)
+        .saturating_add(summary.records_deleted);
+    let message = match override_older_than_days {
+        Some(days) => format!("请求记录手动清理完成，清理 {days} 天前的记录，影响 {total} 项"),
+        None => format!("请求记录手动清理完成（按当前策略），影响 {total} 项"),
+    };
+    record_completed_cleanup_run(
+        data,
+        USAGE_CLEANUP_KIND,
+        "manual",
+        started_at_unix_secs,
+        started_at,
+        json!({
+            "body_externalized": summary.body_externalized,
+            "legacy_body_refs_migrated": summary.legacy_body_refs_migrated,
+            "body_cleaned": summary.body_cleaned,
+            "header_cleaned": summary.header_cleaned,
+            "keys_cleaned": summary.keys_cleaned,
+            "records_deleted": summary.records_deleted,
+            "requested_older_than_days": override_older_than_days,
+            "actor_user_id": actor_user_id,
+        }),
+        message,
+    )
+    .await;
+    info!(
+        event_name = "usage_cleanup_manual_completed",
+        log_type = "ops",
+        worker = "usage_cleanup",
+        trigger = "manual",
+        requested_older_than_days = override_older_than_days,
+        actor_user_id = actor_user_id.as_deref(),
+        total_affected = total,
+        "gateway finished manual usage cleanup"
+    );
+    Ok(summary)
+}
+
+#[derive(Debug)]
+pub(crate) enum ManualUsageCleanupError {
+    AlreadyRunning,
+    DataLayer(DataLayerError),
+}
+
+impl std::fmt::Display for ManualUsageCleanupError {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        match self {
+            Self::AlreadyRunning => f.write_str("a usage cleanup run is already in progress"),
+            Self::DataLayer(err) => write!(f, "{err}"),
+        }
+    }
+}
+
+impl std::error::Error for ManualUsageCleanupError {
+    fn source(&self) -> Option<&(dyn std::error::Error + 'static)> {
+        match self {
+            Self::AlreadyRunning => None,
+            Self::DataLayer(err) => Some(err),
+        }
+    }
+}
+
 pub(super) fn run_pool_monitor_once(data: &GatewayDataState) {
    let Some(summary) = summarize_database_pool(data) else {
        return;
--- a/apps/aether-gateway/src/maintenance/runtime/tests.rs
+++ b/apps/aether-gateway/src/maintenance/runtime/tests.rs
@@ -28,7 +28,8 @@ use super::{
    spawn_usage_cleanup_worker, spawn_wallet_daily_usage_aggregation_worker,
    start_proxy_upgrade_rollout, stats_aggregation_target_day,
    stats_hourly_aggregation_target_hour, summarize_database_pool, usage_cleanup_settings,
-    usage_cleanup_window, wallet_daily_usage_aggregation_target, AppState, DbMaintenanceRunSummary,
+    usage_cleanup_window, usage_cleanup_window_with_override,
+    wallet_daily_usage_aggregation_target, AppState, DbMaintenanceRunSummary,
    FailedPendingUsageRow, GatewayDataState, ProxyNodeMetricsCleanupSettings,
    ProxyUpgradeRolloutProbeConfig, StalePendingUsageRow, UsageCleanupSettings, USAGE_CLEANUP_HOUR,
    USAGE_CLEANUP_MINUTE, WALLET_DAILY_USAGE_AGGREGATION_HOUR,
@@ -940,6 +941,42 @@ fn usage_cleanup_window_uses_non_overlapping_ranges() {
    assert!(window.compressed_cutoff > window.log_cutoff);
 }

+#[test]
+fn usage_cleanup_window_with_override_is_always_non_aggressive() {
+    let now_utc = "2026-03-18T03:00:00Z"
+        .parse::<DateTime<Utc>>()
+        .expect("timestamp should parse");
+    let settings = UsageCleanupSettings {
+        detail_retention_days: 7,
+        compressed_retention_days: 30,
+        header_retention_days: 90,
+        log_retention_days: 365,
+        batch_size: 123,
+        auto_delete_expired_keys: false,
+    };
+    let policy = usage_cleanup_window(now_utc, settings);
+
+    let override_duration = chrono::Duration::days(180);
+    let clamped = usage_cleanup_window_with_override(now_utc, settings, Some(override_duration));
+
+    assert_eq!(clamped.detail_cutoff, policy.detail_cutoff);
+    assert_eq!(clamped.compressed_cutoff, policy.compressed_cutoff);
+    assert_eq!(clamped.header_cutoff, policy.header_cutoff);
+    assert_eq!(clamped.log_cutoff, now_utc - override_duration);
+    assert!(clamped.log_cutoff > policy.log_cutoff);
+
+    let far_override = chrono::Duration::days(5);
+    let far = usage_cleanup_window_with_override(now_utc, settings, Some(far_override));
+    assert_eq!(far.detail_cutoff, now_utc - far_override);
+    assert_eq!(far.compressed_cutoff, now_utc - far_override);
+    assert_eq!(far.header_cutoff, now_utc - far_override);
+    assert_eq!(far.log_cutoff, now_utc - far_override);
+    assert!(far.log_cutoff > policy.log_cutoff);
+
+    let passthrough = usage_cleanup_window_with_override(now_utc, settings, None);
+    assert_eq!(passthrough, policy);
+}
+
 #[tokio::test]
 async fn summarize_database_pool_uses_busy_connections_for_usage_rate() {
    let data = GatewayDataState::from_config(crate::data::GatewayDataConfig::from_postgres_config(
--- a/apps/aether-gateway/src/maintenance/runtime/usage_cleanup.rs
+++ b/apps/aether-gateway/src/maintenance/runtime/usage_cleanup.rs
@@ -1,23 +1,48 @@
-use aether_data_contracts::repository::usage::UsageCleanupSummary;
+use aether_data_contracts::repository::usage::{UsageCleanupSummary, UsageCleanupWindow};
 use aether_data_contracts::DataLayerError;
 use chrono::Utc;

 use crate::data::GatewayDataState;

-use super::{system_config_bool, usage_cleanup_settings, usage_cleanup_window};
+use super::{
+    system_config_bool, usage_cleanup_settings, usage_cleanup_window,
+    usage_cleanup_window_with_override,
+};
+
+#[derive(Debug, Clone, Copy, serde::Serialize)]
+pub(crate) struct ManualUsageCleanupPreview {
+    pub detail_cutoff: chrono::DateTime<Utc>,
+    pub compressed_cutoff: chrono::DateTime<Utc>,
+    pub header_cutoff: chrono::DateTime<Utc>,
+    pub log_cutoff: chrono::DateTime<Utc>,
+    pub requested_older_than_days: Option<u32>,
+    pub detail_count: u64,
+    pub compressed_count: u64,
+    pub header_count: u64,
+    pub log_count: u64,
+}

 pub(super) async fn perform_usage_cleanup_once(
    data: &GatewayDataState,
+) -> Result<UsageCleanupSummary, DataLayerError> {
+    perform_usage_cleanup_once_with_override(data, None).await
+}
+
+pub(super) async fn perform_usage_cleanup_once_with_override(
+    data: &GatewayDataState,
+    override_older_than: Option<chrono::Duration>,
 ) -> Result<UsageCleanupSummary, DataLayerError> {
    if !data.has_usage_writer() {
        return Ok(UsageCleanupSummary::default());
    }
-    if !system_config_bool(data, "enable_auto_cleanup", true).await? {
+    if override_older_than.is_none()
+        && !system_config_bool(data, "enable_auto_cleanup", true).await?
+    {
        return Ok(UsageCleanupSummary::default());
    }

+    let window = compute_usage_cleanup_window(data, override_older_than).await?;
    let settings = usage_cleanup_settings(data).await?;
-    let window = usage_cleanup_window(Utc::now(), settings);
    data.cleanup_usage(
        &window,
        settings.batch_size,
@@ -25,3 +50,35 @@ pub(super) async fn perform_usage_cleanup_once(
    )
    .await
 }
+
+pub(crate) async fn preview_manual_usage_cleanup(
+    data: &GatewayDataState,
+    override_older_than_days: Option<u32>,
+) -> Result<ManualUsageCleanupPreview, DataLayerError> {
+    let override_duration =
+        override_older_than_days.map(|days| chrono::Duration::days(i64::from(days)));
+    let window = compute_usage_cleanup_window(data, override_duration).await?;
+    let counts = data.preview_usage_cleanup(&window).await?;
+    Ok(ManualUsageCleanupPreview {
+        detail_cutoff: window.detail_cutoff,
+        compressed_cutoff: window.compressed_cutoff,
+        header_cutoff: window.header_cutoff,
+        log_cutoff: window.log_cutoff,
+        requested_older_than_days: override_older_than_days,
+        detail_count: counts.detail,
+        compressed_count: counts.compressed,
+        header_count: counts.header,
+        log_count: counts.log,
+    })
+}
+
+async fn compute_usage_cleanup_window(
+    data: &GatewayDataState,
+    override_older_than: Option<chrono::Duration>,
+) -> Result<UsageCleanupWindow, DataLayerError> {
+    let settings = usage_cleanup_settings(data).await?;
+    Ok(match override_older_than {
+        Some(duration) => usage_cleanup_window_with_override(Utc::now(), settings, Some(duration)),
+        None => usage_cleanup_window(Utc::now(), settings),
+    })
+}