yml2213andCursor
647fdc0b4a
利润不再扣支付通道费；站点可配置结算冷却期，到期自动入账，已释放后退款从收款站钱包回冲。同时修复当日自动批次无法追加、三方回冲打错钱包，以及测试库自增重置被 MySQL 统计缓存跳过的问题。 Co-authored-by: Cursor <cursoragent@cursor.com>
118 lines
2.8 KiB
Go
118 lines
2.8 KiB
Go
package distributionsettlement
|
|
|
|
import (
|
|
"context"
|
|
"crypto/rand"
|
|
"encoding/hex"
|
|
"fmt"
|
|
"time"
|
|
|
|
"hfb_sys/backend/internal/modules/distribution"
|
|
|
|
"github.com/redis/go-redis/v9"
|
|
"go.uber.org/zap"
|
|
"gorm.io/gorm"
|
|
)
|
|
|
|
const releaseLockKey = "hfb:job:distributionsettlement:lock"
|
|
|
|
type Job struct {
|
|
db *gorm.DB
|
|
redis *redis.Client
|
|
logger *zap.Logger
|
|
interval time.Duration
|
|
instanceID string
|
|
}
|
|
|
|
func New(db *gorm.DB, redisClient *redis.Client, logger *zap.Logger) *Job {
|
|
return &Job{
|
|
db: db,
|
|
redis: redisClient,
|
|
logger: logger,
|
|
interval: 10 * time.Minute,
|
|
instanceID: newInstanceID(),
|
|
}
|
|
}
|
|
|
|
// acquireLock 通过 Redis 分布式锁确保同一时刻只有一个实例执行结算释放。
|
|
// 未配置 Redis 时直接执行；释放逻辑本身幂等（行锁 + 结算明细唯一性）。
|
|
func (j *Job) acquireLock(ctx context.Context) (func(), bool) {
|
|
if j.redis == nil {
|
|
return func() {}, true
|
|
}
|
|
ok, err := j.redis.SetNX(ctx, releaseLockKey, j.instanceID, j.lockTTL()).Result()
|
|
if err != nil {
|
|
j.logger.Warn("分销自动结算任务获取锁失败，将以单实例模式执行", zap.Error(err))
|
|
return func() {}, true
|
|
}
|
|
if !ok {
|
|
return nil, false
|
|
}
|
|
return j.releaseLock, true
|
|
}
|
|
|
|
// releaseLock 仅在锁仍归本实例时释放，避免误删其他实例的锁。
|
|
func (j *Job) releaseLock() {
|
|
if j.redis == nil {
|
|
return
|
|
}
|
|
relCtx, cancel := context.WithTimeout(context.Background(), 3*time.Second)
|
|
defer cancel()
|
|
script := redis.NewScript(`if redis.call("get", KEYS[1]) == ARGV[1] then return redis.call("del", KEYS[1]) else return 0 end`)
|
|
if err := script.Run(relCtx, j.redis, []string{releaseLockKey}, j.instanceID).Err(); err != nil {
|
|
j.logger.Warn("分销自动结算任务释放锁失败", zap.Error(err))
|
|
}
|
|
}
|
|
|
|
func (j *Job) lockTTL() time.Duration {
|
|
return 2 * j.interval
|
|
}
|
|
|
|
func newInstanceID() string {
|
|
b := make([]byte, 8)
|
|
if _, err := rand.Read(b); err != nil {
|
|
return fmt.Sprintf("inst-%d", time.Now().UnixNano())
|
|
}
|
|
return hex.EncodeToString(b)
|
|
}
|
|
|
|
func (j *Job) Start(ctx context.Context) {
|
|
if j == nil || j.db == nil {
|
|
return
|
|
}
|
|
go j.loop(ctx)
|
|
}
|
|
|
|
func (j *Job) loop(ctx context.Context) {
|
|
j.run(ctx)
|
|
ticker := time.NewTicker(j.interval)
|
|
defer ticker.Stop()
|
|
for {
|
|
select {
|
|
case <-ctx.Done():
|
|
j.logger.Debug("分销自动结算任务已停止")
|
|
return
|
|
case <-ticker.C:
|
|
j.run(ctx)
|
|
}
|
|
}
|
|
}
|
|
|
|
func (j *Job) run(ctx context.Context) {
|
|
release, ok := j.acquireLock(ctx)
|
|
if !ok {
|
|
return
|
|
}
|
|
defer release()
|
|
|
|
repo := distribution.NewRepository(j.db)
|
|
count, err := repo.AutoReleaseDueSettlements(ctx)
|
|
if err != nil {
|
|
j.logger.Warn("分销自动结算释放失败", zap.Error(err))
|
|
return
|
|
}
|
|
if count > 0 {
|
|
j.logger.Info("分销自动结算释放完成", zap.Int("orders", count))
|
|
}
|
|
}
|