package model import ( "time" "gorm.io/gorm" ) const ( PendingSyncStatusPending = "pending" PendingSyncStatusSettled = "settled" PendingSyncStatusSynced = "synced" PendingSyncStatusFailed = "failed" PendingSyncStatusArchived = "archived" ) // PendingSyncRecord 待同步消费记录 type PendingSyncRecord struct { Id int `json:"id" gorm:"primaryKey"` UserId int `json:"user_id" gorm:"index:idx_user_status"` RemoteUserId int `json:"remote_user_id" gorm:"index"` RequestId string `json:"request_id" gorm:"type:varchar(128);uniqueIndex"` Quota int `json:"quota"` EstimatedQuota int `json:"estimated_quota"` Status string `json:"status" gorm:"default:'pending';index:idx_user_status;index:idx_status_created"` CreatedAt int64 `json:"created_at" gorm:"index:idx_status_created"` SyncedAt int64 `json:"synced_at"` RetryCount int `json:"retry_count" gorm:"default:0"` ErrorMsg string `json:"error_msg"` } func (PendingSyncRecord) TableName() string { return "pending_sync_records" } // GetPendingSyncQuota 获取用户待同步金额总和 func GetPendingSyncQuota(userId int) int { var total int64 DB.Model(&PendingSyncRecord{}). Where("user_id = ? AND status IN ?", userId, []string{PendingSyncStatusPending, PendingSyncStatusSettled, PendingSyncStatusFailed}). Select("COALESCE(SUM(quota), 0)"). Scan(&total) return int(total) } // CreatePendingSyncRecord 创建待同步记录(status=pending) func CreatePendingSyncRecord(userId, remoteUserId int, requestId string, quota, estimatedQuota int) error { record := PendingSyncRecord{ UserId: userId, RemoteUserId: remoteUserId, RequestId: requestId, Quota: quota, EstimatedQuota: estimatedQuota, Status: PendingSyncStatusPending, CreatedAt: time.Now().Unix(), } return DB.Create(&record).Error } // CreateSettledSyncRecord 创建已结算的同步记录(PreConsume 跳过时使用) func CreateSettledSyncRecord(userId, remoteUserId int, requestId string, quota int) error { record := PendingSyncRecord{ UserId: userId, RemoteUserId: remoteUserId, RequestId: requestId, Quota: quota, EstimatedQuota: quota, Status: PendingSyncStatusSettled, CreatedAt: time.Now().Unix(), } return DB.Create(&record).Error } // SettlePendingSyncRecord 将 pending 记录结算为 settled,同时更新 quota // 返回 (是否找到并更新, 错误) func SettlePendingSyncRecord(requestId string, actualQuota int) (bool, error) { result := DB.Model(&PendingSyncRecord{}). Where("request_id = ? AND status = ?", requestId, PendingSyncStatusPending). Updates(map[string]interface{}{ "quota": actualQuota, "status": PendingSyncStatusSettled, }) if result.Error != nil { return false, result.Error } return result.RowsAffected > 0, nil } // GetPendingRecordsForSync 获取待同步记录(只同步 settled 和 failed) func GetPendingRecordsForSync(limit, maxRetry int) []PendingSyncRecord { var records []PendingSyncRecord DB.Where("status IN ? AND retry_count < ?", []string{PendingSyncStatusSettled, PendingSyncStatusFailed}, maxRetry). Limit(limit). Order("created_at asc"). Find(&records) return records } // MarkRecordSynced 标记记录为已同步 func MarkRecordSynced(recordId int) error { return DB.Model(&PendingSyncRecord{}).Where("id = ?", recordId).Updates(map[string]interface{}{ "status": PendingSyncStatusSynced, "synced_at": time.Now().Unix(), }).Error } // MarkRecordFailed 标记记录为失败 func MarkRecordFailed(recordId int, errMsg string) error { return DB.Model(&PendingSyncRecord{}).Where("id = ?", recordId).Updates(map[string]interface{}{ "status": PendingSyncStatusFailed, "retry_count": gorm.Expr("retry_count + 1"), "error_msg": errMsg, }).Error } // ArchiveStalePendingRecords 将超时的 pending 记录标记为 archived(请求崩溃的情况) func ArchiveStalePendingRecords(timeoutSeconds int64) int64 { cutoff := time.Now().Unix() - timeoutSeconds result := DB.Model(&PendingSyncRecord{}). Where("status = ? AND created_at < ?", PendingSyncStatusPending, cutoff). Update("status", PendingSyncStatusArchived) if result.Error != nil { return 0 } return result.RowsAffected } // CleanupOldSyncedRecords 清理超过指定天数的已同步记录 func CleanupOldSyncedRecords(days int) int64 { result := DB.Where("status = ? AND synced_at < ?", PendingSyncStatusSynced, time.Now().Unix()-int64(days*86400)). Delete(&PendingSyncRecord{}) return result.RowsAffected } // GetPendingQuotaByUser 获取每个用户的待同步金额 func GetPendingQuotaByUser() map[int]int { type result struct { UserId int Total int } var results []result DB.Model(&PendingSyncRecord{}). Select("user_id, COALESCE(SUM(quota), 0) as total"). Where("status IN ?", []string{PendingSyncStatusPending, PendingSyncStatusSettled, PendingSyncStatusFailed}). Group("user_id"). Scan(&results) m := make(map[int]int) for _, r := range results { m[r.UserId] = r.Total } return m }