feat(distribution): 通道费改主站承担,并按冷却期自动释放结算
利润不再扣支付通道费;站点可配置结算冷却期,到期自动入账,已释放后退款从收款站钱包回冲。同时修复当日自动批次无法追加、三方回冲打错钱包,以及测试库自增重置被 MySQL 统计缓存跳过的问题。 Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
@@ -0,0 +1,117 @@
|
||||
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))
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user