feat(storage): archive expired usage events

.env.example

@@ -116,10 +116,6 @@ LOG_FILE_ENABLED=true
 # Number of historical days to retain combined logs, plus the current day; 0 disables cleanup; error-only logs keep 30 history days plus the current day. Required: no. Default: 7.
 LOG_RETENTION_DAYS=7
 
-# 是否在每日维护中删除过期 usage_events 原始事件；默认不清理，设为 true 后会删除早于 90 个本地自然日前 00:00 的数据。必填：否。默认值：false。
-# Whether daily maintenance deletes expired raw usage_events; disabled by default. When true, rows earlier than local midnight 90 calendar days ago are deleted. Required: no. Default: false.
-CLEANUP_USAGE_EVENTS_ENABLED=false
-
 # 是否启用 SQLite 数据库备份。必填：否。默认值：true。
 # Whether to enable SQLite database backups. Required: no. Default: true.
 BACKUP_ENABLED=true

README.md

@@ -416,11 +416,12 @@ Scheduled Auth Files quota refresh is configured from the gear button in the Aut
 | `LOG_LEVEL` | No | `info` | Log level |
 | `LOG_FILE_ENABLED` | No | `true` | Write persistent log files |
 | `LOG_RETENTION_DAYS` | No | `7` | Combined-log history days, plus the current day; `0` disables cleanup. Error-only logs keep 30 history days plus the current day |
-| `CLEANUP_USAGE_EVENTS_ENABLED` | No | `false` | Delete expired raw `usage_events` during daily maintenance; when enabled, rows earlier than local midnight 90 calendar days ago are deleted |
 | `BACKUP_ENABLED` | No | `true` | Enable SQLite database backups |
 | `BACKUP_INTERVAL` | No | `24h` | Database backup interval |
 | `BACKUP_RETENTION_DAYS` | No | `7` | Backup retention days |
 
+Keeper automatically moves raw `usage_events` older than 90 local calendar days into the permanently retained `usage_events_archive` cold table during the daily 04:30 maintenance window. The archive is reserved for future schema-migration rebuilds and is not queried by normal dashboard APIs.
+
 When file logging is enabled, `cpa-usage-keeper-YYYY-MM-DD.log` contains all emitted levels. Error, fatal, and panic entries are also copied to `cpa-usage-keeper-error-YYYY-MM-DD.log`, which keeps the previous 30 local calendar dates plus the current date.
 
 ### Built-In HTTPS

README.zh.md

@@ -416,11 +416,12 @@ Auth Files 定时限额刷新在 Auth Files 巡检弹窗的小齿轮中配置。
 | `LOG_LEVEL` | 否 | `info` | 日志级别 |
 | `LOG_FILE_ENABLED` | 否 | `true` | 是否写入持久化日志文件 |
 | `LOG_RETENTION_DAYS` | 否 | `7` | 综合日志保留历史天数，并额外保留当天；`0` 表示不自动清理。仅错误日志固定保留历史 30 天及当天 |
-| `CLEANUP_USAGE_EVENTS_ENABLED` | 否 | `false` | 是否在每日维护中删除过期 `usage_events` 原始事件；启用后会删除早于 90 个本地自然日前 00:00 的数据 |
 | `BACKUP_ENABLED` | 否 | `true` | 是否启用 SQLite 数据库备份 |
 | `BACKUP_INTERVAL` | 否 | `24h` | 数据库备份间隔 |
 | `BACKUP_RETENTION_DAYS` | 否 | `7` | 备份保留天数 |
 
+Keeper 会在每天 04:30 的维护窗口中，把早于 90 个本地自然日的原始 `usage_events` 自动移动到永久保留的 `usage_events_archive` 冷表。该冷表用于未来 schema migration 重建增量数据，正常仪表盘 API 不查询 archive。
+
 启用文件日志后，`cpa-usage-keeper-YYYY-MM-DD.log` 会记录所有已输出级别；error、fatal 和 panic 级别还会同时写入 `cpa-usage-keeper-error-YYYY-MM-DD.log`，该文件固定保留历史 30 个本地自然日及当天。
 
 ### 内置 HTTPS

go.mod

@@ -7,6 +7,7 @@ require (
 	github.com/joho/godotenv v1.5.1
 	github.com/mattn/go-sqlite3 v1.14.48
 	github.com/sirupsen/logrus v1.9.3
+	golang.org/x/sys v0.20.0
 	golang.org/x/text v0.20.0
 	gorm.io/driver/sqlite v1.5.7
 	gorm.io/gorm v1.26.1
@@ -38,7 +39,6 @@ require (
 	golang.org/x/arch v0.8.0 // indirect
 	golang.org/x/crypto v0.23.0 // indirect
 	golang.org/x/net v0.25.0 // indirect
-	golang.org/x/sys v0.20.0 // indirect
 	google.golang.org/protobuf v1.34.1 // indirect
 	gopkg.in/yaml.v3 v3.0.1 // indirect
 )

internal/app/app.go

@@ -181,9 +181,8 @@ func NewWithConfig(cfg config.Config) (*App, error) {
 	usageAggregationRunner := poller.NewUsageAggregationRunner(db)
 	// syncService 仍然是 metadata 和 usage 处理共享的业务服务入口。
 	syncService := service.NewSyncServiceWithOptions(db, service.SyncServiceOptions{
-		BaseURL:                   cfg.CPABaseURL,
-		Client:                    cpaClient,
-		CleanupUsageEventsEnabled: cfg.CleanupUsageEventsEnabled,
+		BaseURL: cfg.CPABaseURL,
+		Client:  cpaClient,
 		// usage_events 事务提交后通过这个缓存做非阻塞增量追加，供 Overview realtime 和右边界补偿复用。
 		RecentUsageEvents: recentUsageCache,
 		// usage 与 metadata 提交后只唤醒单 writer runner，不在前台链路执行派生聚合。

internal/config/config.go

@@ -78,8 +78,6 @@ type Config struct {
 	BackupInterval time.Duration
 	// BackupRetentionDays 是备份文件保留天数。
 	BackupRetentionDays int
-	// CleanupUsageEventsEnabled 控制每日维护是否删除过期 usage_events 原始事件。
-	CleanupUsageEventsEnabled bool
 	// RequestTimeout 是访问 CPA HTTP 和 Redis TCP 的超时时间。
 	RequestTimeout time.Duration
 	// TLSSkipVerify 控制是否跳过 CPA HTTPS 和 Redis 队列 TLS 的证书验证。
@@ -185,11 +183,6 @@ func Load(options LoadOptions) (*Config, error) {
 	if backupRetentionDays < 0 {
 		return nil, fmt.Errorf("BACKUP_RETENTION_DAYS must be non-negative")
 	}
-	cleanupUsageEventsEnabled, err := getBool("CLEANUP_USAGE_EVENTS_ENABLED", false)
-	if err != nil {
-		return nil, err
-	}
-
 	logFileEnabled, err := getBool("LOG_FILE_ENABLED", true)
 	if err != nil {
 		return nil, err
@@ -263,7 +256,6 @@ func Load(options LoadOptions) (*Config, error) {
 		BackupDir:                  filepath.Join(workDir, workDirBackupsName),
 		BackupInterval:             backupInterval,
 		BackupRetentionDays:        backupRetentionDays,
-		CleanupUsageEventsEnabled:  cleanupUsageEventsEnabled,
 		RequestTimeout:             requestTimeout,
 		TLSSkipVerify:              tlsSkipVerify,
 		LogLevel:                   getString("LOG_LEVEL", "info"),

internal/config/config_test.go

@@ -15,7 +15,7 @@ import (
 var configEnvKeys = []string{
 	"APP_HOST", "APP_PORT", "APP_BASE_PATH", "CPA_PUBLIC_URL", "WORK_DIR", "CPA_BASE_URL", "CPA_MANAGEMENT_KEY", "POLL_INTERVAL",
 	"USAGE_SYNC_MODE", "REDIS_QUEUE_ADDR", "REDIS_QUEUE_TLS", "REDIS_QUEUE_BATCH_SIZE", "REDIS_QUEUE_IDLE_INTERVAL",
-	"SQLITE_PATH", "BACKUP_ENABLED", "BACKUP_DIR", "BACKUP_INTERVAL", "BACKUP_RETENTION_DAYS", "CLEANUP_USAGE_EVENTS_ENABLED",
+	"SQLITE_PATH", "BACKUP_ENABLED", "BACKUP_DIR", "BACKUP_INTERVAL", "BACKUP_RETENTION_DAYS",
 	"REQUEST_TIMEOUT", "LOG_LEVEL", "LOG_FILE_ENABLED", "LOG_DIR", "LOG_RETENTION_DAYS",
 	"AUTH_ENABLED", "LOGIN_PASSWORD", "AUTH_SESSION_TTL", "TZ", "TLS_SKIP_VERIFY", "QUOTA_REFRESH_WORKER_LIMIT",
 }

internal/config/test/config_cleanup_test.go

@@ -13,43 +13,12 @@ var isolatedConfigEnvKeys = []string{
 	"APP_HOST", "APP_PORT", "APP_BASE_PATH", "CPA_PUBLIC_URL", "WORK_DIR", "CPA_BASE_URL", "CPA_MANAGEMENT_KEY",
 	"CPA_REQUEST_LOG_ACCESS_ENABLED",
 	"REDIS_QUEUE_ADDR", "REDIS_QUEUE_TLS", "REDIS_QUEUE_BATCH_SIZE", "REDIS_QUEUE_IDLE_INTERVAL",
-	"BACKUP_ENABLED", "BACKUP_INTERVAL", "BACKUP_RETENTION_DAYS", "CLEANUP_USAGE_EVENTS_ENABLED",
+	"BACKUP_ENABLED", "BACKUP_INTERVAL", "BACKUP_RETENTION_DAYS",
 	"REQUEST_TIMEOUT", "LOG_LEVEL", "LOG_FILE_ENABLED", "LOG_DIR", "LOG_RETENTION_DAYS",
 	"AUTH_ENABLED", "LOGIN_PASSWORD", "AUTH_SESSION_TTL", "TZ", "TLS_ENABLED", "TLS_CERT_FILE", "TLS_KEY_FILE",
 	"TLS_SKIP_VERIFY", "QUOTA_REFRESH_WORKER_LIMIT",
 }
 
-func TestLoadFromEnvDefaultsUsageEventCleanupDisabled(t *testing.T) {
-	isolateConfigEnv(t)
-	t.Setenv("CPA_BASE_URL", "http://127.0.0.1:"+cpa.ManagementRedisDefaultPort)
-	t.Setenv("CPA_MANAGEMENT_KEY", "secret")
-
-	cfg, err := config.LoadFromEnv()
-	if err != nil {
-		t.Fatalf("LoadFromEnv returned error: %v", err)
-	}
-
-	if cfg.CleanupUsageEventsEnabled {
-		t.Fatal("expected usage_events cleanup to be disabled by default")
-	}
-}
-
-func TestLoadFromEnvReadsUsageEventCleanupFlag(t *testing.T) {
-	isolateConfigEnv(t)
-	t.Setenv("CPA_BASE_URL", "http://127.0.0.1:"+cpa.ManagementRedisDefaultPort)
-	t.Setenv("CPA_MANAGEMENT_KEY", "secret")
-	t.Setenv("CLEANUP_USAGE_EVENTS_ENABLED", "true")
-
-	cfg, err := config.LoadFromEnv()
-	if err != nil {
-		t.Fatalf("LoadFromEnv returned error: %v", err)
-	}
-
-	if !cfg.CleanupUsageEventsEnabled {
-		t.Fatal("expected usage_events cleanup to be enabled")
-	}
-}
-
 func TestLoadFromEnvDefaultsCPARequestLogAccessDisabled(t *testing.T) {
 	isolateConfigEnv(t)
 	t.Setenv("CPA_BASE_URL", "http://127.0.0.1:"+cpa.ManagementRedisDefaultPort)

internal/entities/all.go

@@ -4,6 +4,7 @@ package entities
 func All() []any {
 	return []any{
 		&UsageEvent{},
+		&UsageEventArchive{},
 		&RedisUsageInbox{},
 		&ModelPriceSetting{},
 		&ModelPriceRule{},

internal/entities/test/entities_test.go

@@ -15,6 +15,7 @@ func TestAllIncludesCoreModels(t *testing.T) {
 	items := All()
 	expected := []any{
 		&UsageEvent{},
+		&UsageEventArchive{},
 		&RedisUsageInbox{},
 		&ModelPriceSetting{},
 		&ModelPriceRule{},

internal/entities/usage_event_archive.go

@@ -0,0 +1,46 @@
+package entities
+
+import "time"
+
+// UsageEventStorageColumns 是 hot/archive 原始事件复制使用的完整持久化列契约。
+const UsageEventStorageColumns = "id, event_key, api_group_key, provider, endpoint, auth_type, request_id, client_ip, x_forwarded_for, user_agent, model, model_alias, reasoning_effort, service_tier, response_service_tier, executor_type, timestamp, source, auth_index, failed, generate, latency_ms, ttft_ms, input_tokens, output_tokens, reasoning_tokens, cached_tokens, cache_read_tokens, cache_creation_tokens, total_tokens, created_at"
+
+// UsageEventArchive 永久保存已经离开 hot usage_events 的原始事件。
+// 字段必须与 UsageEvent 的持久化列保持一致，但 archive 不承担在线查询，因此不复制二级索引。
+type UsageEventArchive struct {
+	ID                  int64 `gorm:"primaryKey;autoIncrement:false"`
+	EventKey            string
+	APIGroupKey         string
+	Provider            string  `gorm:"column:provider"`
+	Endpoint            string  `gorm:"column:endpoint"`
+	AuthType            string  `gorm:"column:auth_type"`
+	RequestID           string  `gorm:"column:request_id"`
+	ClientIP            *string `gorm:"column:client_ip"`
+	XForwardedFor       *string `gorm:"column:x_forwarded_for"`
+	UserAgent           *string `gorm:"column:user_agent"`
+	Model               string
+	ModelAlias          *string   `gorm:"column:model_alias"`
+	ReasoningEffort     string    `gorm:"column:reasoning_effort;not null;default:''"`
+	ServiceTier         string    `gorm:"column:service_tier;not null;default:''"`
+	ResponseServiceTier string    `gorm:"column:response_service_tier;not null;default:''"`
+	ExecutorType        string    `gorm:"column:executor_type;not null;default:''"`
+	Timestamp           time.Time `gorm:"serializer:storageTime"`
+	Source              string
+	AuthIndex           string
+	Failed              bool
+	Generate            *bool `gorm:"column:generate;not null;default:true"`
+	LatencyMS           int64
+	TTFTMS              *int64 `gorm:"column:ttft_ms"`
+	InputTokens         int64
+	OutputTokens        int64
+	ReasoningTokens     int64
+	CachedTokens        int64
+	CacheReadTokens     int64 `gorm:"not null;default:0"`
+	CacheCreationTokens int64 `gorm:"not null;default:0"`
+	TotalTokens         int64
+	CreatedAt           time.Time `gorm:"serializer:storageTime"`
+}
+
+func (UsageEventArchive) TableName() string {
+	return "usage_events_archive"
+}

internal/repository/db.go

@@ -335,103 +335,59 @@ func InsertUsageEvents(db *gorm.DB, events []entities.UsageEvent) (int, int, err
 	return inserted, 0, nil
 }
 
-type CleanupStorageOptions struct {
-	// CleanupUsageEvents 控制是否删除过期 usage_events 原始事件；默认 false 表示保留原始事件。
-	CleanupUsageEvents bool
-}
-
 const usageEventsRetentionDays = 90
 
-// CleanupStorage 是每日维护任务的统一仓储清理入口：先清 inbox/raw events，再清 Activity 与 Latency 限期统计，最后执行 VACUUM。
-// VACUUM 必须在删除完成后单独执行，任何一步失败都会停止后续步骤并把已完成部分的结果返回给上层日志。
-func CleanupStorage(db *gorm.DB, now time.Time, options ...CleanupStorageOptions) (dto.StorageCleanupResult, error) {
-	opts := CleanupStorageOptions{}
-	if len(options) > 0 {
-		opts = options[0]
-	}
+// CleanupStorage 是每日维护任务的统一仓储入口：先清 inbox、归档 raw events，再清限期统计并条件式整理空闲页。
+func CleanupStorage(db *gorm.DB, now time.Time) (dto.StorageCleanupResult, error) {
 	redisResult, err := CleanupRedisUsageInbox(db, now)
 	if err != nil {
 		return dto.StorageCleanupResult{RedisInbox: redisResult}, err
 	}
-	var usageEventsDeleted int64
-	if opts.CleanupUsageEvents {
-		usageEventsDeleted, err = cleanupUsageEvents(db, now)
-		if err != nil {
-			return dto.StorageCleanupResult{RedisInbox: redisResult, UsageEventsDeleted: usageEventsDeleted}, err
-		}
+	usageEventsArchive, err := ArchiveExpiredUsageEvents(databaseContext(db), db, now)
+	result := dto.StorageCleanupResult{
+		RedisInbox:               redisResult,
+		UsageEventsArchived:      usageEventsArchive.Archived,
+		UsageEventsArchiveStatus: usageEventsArchive.Status,
+	}
+	if err != nil {
+		return result, err
 	}
 	// Activity 的 short/medium/long 分别按自身 retention 清理，daily 永久保留。
 	if err := CleanupUsageActivityStats(db, now); err != nil {
-		return dto.StorageCleanupResult{RedisInbox: redisResult, UsageEventsDeleted: usageEventsDeleted}, err
+		return result, err
 	}
-	// Latency 小时保留 3 天、自然日保留 365 天，并复用本轮最后一次 VACUUM。
+	// Latency 小时保留 3 天、自然日保留 365 天；空闲页由后面的条件式 VACUUM 统一评估。
 	if err := CleanupUsageLatencyStats(db, now); err != nil {
-		return dto.StorageCleanupResult{RedisInbox: redisResult, UsageEventsDeleted: usageEventsDeleted}, err
-	}
-	// SQLite 删除不会立即缩小文件，维护窗口最后统一 VACUUM。
-	if err := db.Exec("VACUUM").Error; err != nil {
-		return dto.StorageCleanupResult{RedisInbox: redisResult, UsageEventsDeleted: usageEventsDeleted}, err
+		return result, err
 	}
-	return dto.StorageCleanupResult{RedisInbox: redisResult, UsageEventsDeleted: usageEventsDeleted}, nil
-}
-
-// cleanupUsageEvents 严格删除早于“time.Local 当日零点向前 90 个自然日”边界的原始 usage_events。
-func cleanupUsageEvents(db *gorm.DB, now time.Time) (int64, error) {
-	if db == nil {
-		return 0, fmt.Errorf("database is nil")
-	}
-	// deleted 只在安全水位检查和 DELETE 同一事务提交后返回。
-	deleted := int64(0)
-	// 安全检查与删除共用事务，禁止新 usage event 在两者之间插入并被误删。
-	err := db.Transaction(func(tx *gorm.DB) error {
-		// 三类聚合任一落后时，本轮跳过 raw event 删除并等待下一次维护窗口。
-		safe, err := usageEventAggregationsCaughtUp(tx)
-		if err != nil {
-			return err
-		}
-		// 跳过删除不是维护失败，Activity retention 和 VACUUM 仍可继续执行。
-		if !safe {
-			return nil
-		}
-		// 只有安全水位已经覆盖当前最大 ID 时才计算时间保留线。
-		cutoff := usageEventsCleanupCutoff(now)
-		// DELETE 保持严格小于 cutoff，边界时刻本身必须保留。
-		result := tx.Unscoped().Where("timestamp < ?", timeutil.FormatStorageTime(cutoff)).Delete(&entities.UsageEvent{})
-		if result.Error != nil {
-			return fmt.Errorf("cleanup usage events: %w", result.Error)
-		}
-		// 事务成功返回前记录本次真实删除行数。
-		deleted = result.RowsAffected
-		return nil
-	})
-	// 事务失败时不报告未提交的删除数量。
+	vacuumResult, err := maybeVacuumStorage(db)
+	result.Vacuum = vacuumResult
 	if err != nil {
-		return 0, err
+		return result, err
 	}
-	// 返回已经提交的删除数量；聚合落后时固定为 0。
-	return deleted, nil
+	return result, nil
 }
 
 func usageEventAggregationsCaughtUp(tx *gorm.DB) (bool, error) {
 	// 当前最大 event ID 是三个全局 checkpoint 和每行 Identity cursor 的共同安全目标。
 	var maxEventID int64
 	if err := tx.Model(&entities.UsageEvent{}).Select("COALESCE(MAX(id), 0)").Scan(&maxEventID).Error; err != nil {
-		return false, fmt.Errorf("load usage event cleanup watermark: %w", err)
+		return false, fmt.Errorf("load usage event archive watermark: %w", err)
 	}
-	// 空 raw event 表没有待聚合数据，可以直接执行空删除。
+	// 空 hot 表没有待聚合数据，也没有需要归档的事件。
 	if maxEventID == 0 {
 		return true, nil
 	}
 
-	// 同一事务内一次读取三行，避免分别查询时观察到不一致水位或继续依赖已删除旧表。
+	// 同一事务内一次读取三行，避免分别查询时观察到不一致水位。
 	var checkpoints []entities.UsageAggregationCheckpoint
 	names := []entities.UsageAggregationCheckpointName{
 		entities.UsageAggregationCheckpointOverview,
 		entities.UsageAggregationCheckpointActivity,
 		entities.UsageAggregationCheckpointLatency,
 	}
 	if err := tx.Where("name IN ?", names).Find(&checkpoints).Error; err != nil {
-		return false, fmt.Errorf("load usage aggregation cleanup watermarks: %w", err)
+		return false, fmt.Errorf("load usage aggregation archive watermarks: %w", err)
 	}
 	// 缺行、重复异常或任一 cursor 落后都必须保守保留 raw events。
 	ready := make(map[entities.UsageAggregationCheckpointName]bool, len(names))
@@ -444,16 +400,16 @@ func usageEventAggregationsCaughtUp(tx *gorm.DB) (bool, error) {
 		}
 	}
 
-	// Identity 没有全局 checkpoint；复用兼容追赶的同一 EXISTS 判断，但保持在当前 cleanup 事务内读取。
+	// Identity 没有全局 checkpoint；复用兼容追赶的同一 EXISTS 判断，但保持在当前 archive 事务内读取。
 	pendingIdentity, err := hasPendingUsageIdentityAggregation(tx)
 	if err != nil {
-		return false, fmt.Errorf("check identity cleanup watermark: %w", err)
+		return false, fmt.Errorf("check identity archive watermark: %w", err)
 	}
-	// 任一 active/deleted identity 仍有 delta 时都禁止删除；不存在匹配 delta 才算安全。
+	// 任一 active/deleted identity 仍有 delta 时都禁止事件离开 hot 表。
 	return !pendingIdentity, nil
 }
 
-func usageEventsCleanupCutoff(now time.Time) time.Time {
+func usageEventsArchiveCutoff(now time.Time) time.Time {
 	localNow := now.In(time.Local)
 	localDayStart := time.Date(localNow.Year(), localNow.Month(), localNow.Day(), 0, 0, 0, 0, time.Local)
 	return localDayStart.AddDate(0, 0, -usageEventsRetentionDays)

internal/repository/disk_space_unix.go

@@ -0,0 +1,20 @@
+//go:build !windows
+
+package repository
+
+import (
+	"fmt"
+
+	"golang.org/x/sys/unix"
+)
+
+func storageAvailableDiskBytes(path string) (uint64, error) {
+	var stats unix.Statfs_t
+	if err := unix.Statfs(path, &stats); err != nil {
+		return 0, fmt.Errorf("load available disk space for %s: %w", path, err)
+	}
+	if stats.Bsize <= 0 {
+		return 0, fmt.Errorf("load available disk space for %s: invalid block size %d", path, stats.Bsize)
+	}
+	return uint64(stats.Bavail) * uint64(stats.Bsize), nil
+}