|
- 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
- }
|