Nelze vybrat více než 25 témat Téma musí začínat písmenem nebo číslem, může obsahovat pomlčky („-“) a může být dlouhé až 35 znaků.
 
 
 

161 řádky
5.1 KiB

  1. package model
  2. import (
  3. "time"
  4. "gorm.io/gorm"
  5. )
  6. const (
  7. PendingSyncStatusPending = "pending"
  8. PendingSyncStatusSettled = "settled"
  9. PendingSyncStatusSynced = "synced"
  10. PendingSyncStatusFailed = "failed"
  11. PendingSyncStatusArchived = "archived"
  12. )
  13. // PendingSyncRecord 待同步消费记录
  14. type PendingSyncRecord struct {
  15. Id int `json:"id" gorm:"primaryKey"`
  16. UserId int `json:"user_id" gorm:"index:idx_user_status"`
  17. RemoteUserId int `json:"remote_user_id" gorm:"index"`
  18. RequestId string `json:"request_id" gorm:"type:varchar(128);uniqueIndex"`
  19. Quota int `json:"quota"`
  20. EstimatedQuota int `json:"estimated_quota"`
  21. Status string `json:"status" gorm:"default:'pending';index:idx_user_status;index:idx_status_created"`
  22. CreatedAt int64 `json:"created_at" gorm:"index:idx_status_created"`
  23. SyncedAt int64 `json:"synced_at"`
  24. RetryCount int `json:"retry_count" gorm:"default:0"`
  25. ErrorMsg string `json:"error_msg"`
  26. }
  27. func (PendingSyncRecord) TableName() string {
  28. return "pending_sync_records"
  29. }
  30. // GetPendingSyncQuota 获取用户待同步金额总和
  31. func GetPendingSyncQuota(userId int) int {
  32. var total int64
  33. DB.Model(&PendingSyncRecord{}).
  34. Where("user_id = ? AND status IN ?", userId, []string{PendingSyncStatusPending, PendingSyncStatusSettled, PendingSyncStatusFailed}).
  35. Select("COALESCE(SUM(quota), 0)").
  36. Scan(&total)
  37. return int(total)
  38. }
  39. // CreatePendingSyncRecord 创建待同步记录(status=pending)
  40. func CreatePendingSyncRecord(userId, remoteUserId int, requestId string, quota, estimatedQuota int) error {
  41. record := PendingSyncRecord{
  42. UserId: userId,
  43. RemoteUserId: remoteUserId,
  44. RequestId: requestId,
  45. Quota: quota,
  46. EstimatedQuota: estimatedQuota,
  47. Status: PendingSyncStatusPending,
  48. CreatedAt: time.Now().Unix(),
  49. }
  50. return DB.Create(&record).Error
  51. }
  52. // CreateSettledSyncRecord 创建已结算的同步记录(PreConsume 跳过时使用)
  53. func CreateSettledSyncRecord(userId, remoteUserId int, requestId string, quota int) error {
  54. record := PendingSyncRecord{
  55. UserId: userId,
  56. RemoteUserId: remoteUserId,
  57. RequestId: requestId,
  58. Quota: quota,
  59. EstimatedQuota: quota,
  60. Status: PendingSyncStatusSettled,
  61. CreatedAt: time.Now().Unix(),
  62. }
  63. return DB.Create(&record).Error
  64. }
  65. // SettlePendingSyncRecord 将 pending 记录结算为 settled,同时更新 quota
  66. // 返回 (是否找到并更新, 错误)
  67. func SettlePendingSyncRecord(requestId string, actualQuota int) (bool, error) {
  68. result := DB.Model(&PendingSyncRecord{}).
  69. Where("request_id = ? AND status = ?", requestId, PendingSyncStatusPending).
  70. Updates(map[string]interface{}{
  71. "quota": actualQuota,
  72. "status": PendingSyncStatusSettled,
  73. })
  74. if result.Error != nil {
  75. return false, result.Error
  76. }
  77. return result.RowsAffected > 0, nil
  78. }
  79. // GetPendingRecordsForSync 获取待同步记录(只同步 settled 和 failed)
  80. // 优先同步 quota > 0 的记录,quota=0 的旧记录不会阻塞正常扣费同步
  81. func GetPendingRecordsForSync(limit, maxRetry int) []PendingSyncRecord {
  82. var records []PendingSyncRecord
  83. DB.Where("status IN ? AND retry_count < ? AND quota > 0",
  84. []string{PendingSyncStatusSettled, PendingSyncStatusFailed},
  85. maxRetry).
  86. Limit(limit).
  87. Order("created_at asc").
  88. Find(&records)
  89. return records
  90. }
  91. // MarkRecordSynced 标记记录为已同步
  92. func MarkRecordSynced(recordId int) error {
  93. return DB.Model(&PendingSyncRecord{}).Where("id = ?", recordId).Updates(map[string]interface{}{
  94. "status": PendingSyncStatusSynced,
  95. "synced_at": time.Now().Unix(),
  96. }).Error
  97. }
  98. // MarkRecordFailed 标记记录为失败
  99. func MarkRecordFailed(recordId int, errMsg string) error {
  100. return DB.Model(&PendingSyncRecord{}).Where("id = ?", recordId).Updates(map[string]interface{}{
  101. "status": PendingSyncStatusFailed,
  102. "retry_count": gorm.Expr("retry_count + 1"),
  103. "error_msg": errMsg,
  104. }).Error
  105. }
  106. // ArchiveStalePendingRecords 将超时的 pending 记录标记为 archived(请求崩溃的情况)
  107. func ArchiveStalePendingRecords(timeoutSeconds int64) int64 {
  108. cutoff := time.Now().Unix() - timeoutSeconds
  109. result := DB.Model(&PendingSyncRecord{}).
  110. Where("status = ? AND created_at < ?", PendingSyncStatusPending, cutoff).
  111. Update("status", PendingSyncStatusArchived)
  112. if result.Error != nil {
  113. return 0
  114. }
  115. return result.RowsAffected
  116. }
  117. // CleanupOldSyncedRecords 清理超过指定天数的已同步记录
  118. func CleanupOldSyncedRecords(days int) int64 {
  119. result := DB.Where("status = ? AND synced_at < ?",
  120. PendingSyncStatusSynced,
  121. time.Now().Unix()-int64(days*86400)).
  122. Delete(&PendingSyncRecord{})
  123. return result.RowsAffected
  124. }
  125. // GetPendingQuotaByUser 获取每个用户的待同步金额
  126. func GetPendingQuotaByUser() map[int]int {
  127. type result struct {
  128. UserId int
  129. Total int
  130. }
  131. var results []result
  132. DB.Model(&PendingSyncRecord{}).
  133. Select("user_id, COALESCE(SUM(quota), 0) as total").
  134. Where("status IN ?", []string{PendingSyncStatusPending, PendingSyncStatusSettled, PendingSyncStatusFailed}).
  135. Group("user_id").
  136. Scan(&results)
  137. m := make(map[int]int)
  138. for _, r := range results {
  139. m[r.UserId] = r.Total
  140. }
  141. return m
  142. }