389 lines
17 KiB
Go
389 lines
17 KiB
Go
package mysql
|
||
|
||
import (
|
||
"context"
|
||
"database/sql"
|
||
"errors"
|
||
"strconv"
|
||
"strings"
|
||
"time"
|
||
|
||
"hyapp/pkg/appcode"
|
||
"hyapp/pkg/xerr"
|
||
"hyapp/services/wallet-service/internal/domain/ledger"
|
||
)
|
||
|
||
const (
|
||
pointWithdrawalChannelCoinSeller = "coin_seller"
|
||
pointWithdrawalChannelPlatform = "platform"
|
||
|
||
withdrawalLimitPeriodDay = "day"
|
||
withdrawalLimitPeriodWeek = "week"
|
||
withdrawalLimitPeriodMonth = "month"
|
||
)
|
||
|
||
type hostPolicyWithdrawalLimit struct {
|
||
PolicyID uint64
|
||
PolicyVersion uint64
|
||
Period string
|
||
Limit int64
|
||
PeriodKey string
|
||
AllowedDays string
|
||
}
|
||
|
||
type missingHostPolicyMode uint8
|
||
|
||
const (
|
||
// 默认值必须失败关闭，避免新增调用点遗漏参数时意外绕过 POINT 的政策约束。
|
||
requireHostPolicy missingHostPolicyMode = iota
|
||
// Host/Agency 旧工资余额可能早于政策发布；缺少本月及紧邻上月政策时继续沿用原无限次规则。
|
||
allowLegacyWithoutHostPolicy
|
||
)
|
||
|
||
// consumeHostPolicyWithdrawalLimit 在真实账变事务内解析当前 UTC 月绑定政策并锁定一条计数主键。
|
||
// 调用方必须先完成 wallet transaction 幂等命中检查；这样重放直接返回旧回执，不会重复消耗次数。
|
||
// consumeHostPolicyWithdrawalLimitForPolicyType 允许永久积分余额继续使用其最后一份已发布
|
||
// POINT_DIAMOND 快照中的限制；政策切回工资型后，旧积分不能因为当前月类型改变而失去退出能力。
|
||
func (r *Repository) consumeHostPolicyWithdrawalLimitForPolicyType(ctx context.Context, tx *sql.Tx, userID int64, regionID int64, channel string, commandID string, withdrawalRef string, nowMS int64, sourceAssetType string, policyType string, missingPolicyMode missingHostPolicyMode) error {
|
||
if userID <= 0 || regionID < 0 || (channel != pointWithdrawalChannelCoinSeller && channel != pointWithdrawalChannelPlatform) {
|
||
return xerr.New(xerr.InvalidArgument, "point withdrawal limit target is invalid")
|
||
}
|
||
sourceAssetType = strings.ToUpper(strings.TrimSpace(sourceAssetType))
|
||
if sourceAssetType == "" {
|
||
return xerr.New(xerr.InvalidArgument, "withdrawal source asset type is required")
|
||
}
|
||
if missingPolicyMode != requireHostPolicy && missingPolicyMode != allowLegacyWithoutHostPolicy {
|
||
return xerr.New(xerr.Internal, "host policy missing mode is invalid")
|
||
}
|
||
limit, found, err := r.resolveHostPolicyWithdrawalLimitForPolicyType(ctx, tx, regionID, channel, nowMS, policyType)
|
||
if err != nil {
|
||
return err
|
||
}
|
||
if !found && missingPolicyMode == requireHostPolicy {
|
||
// POINT 只有后台政策才能定义提现口径；不存在政策时禁止按工资余额兼容规则放行。
|
||
return xerr.New(xerr.PermissionDenied, "host salary policy is not configured for withdrawal region")
|
||
}
|
||
if channel == pointWithdrawalChannelPlatform && strings.TrimSpace(withdrawalRef) == "" {
|
||
// 平台先冻结再建申请；没有稳定引用就无法在建单失败时精确撤销这一次 reservation。
|
||
return xerr.New(xerr.InvalidArgument, "withdrawal_ref is required for platform withdrawal")
|
||
}
|
||
if found && channel == pointWithdrawalChannelPlatform {
|
||
allowed, err := platformWithdrawalAllowedOnUTCDate(limit.AllowedDays, time.UnixMilli(nowMS).UTC())
|
||
if err != nil {
|
||
return err
|
||
}
|
||
if !allowed {
|
||
return xerr.NewWithMetadata(xerr.PointWithdrawalDateNotAllowed, "point withdrawal date is not allowed", map[string]string{
|
||
"allowed_days": limit.AllowedDays,
|
||
"current_day": strconv.Itoa(time.UnixMilli(nowMS).UTC().Day()),
|
||
"timezone": "UTC",
|
||
})
|
||
}
|
||
}
|
||
|
||
appCode := appcode.FromContext(ctx)
|
||
if found && limit.Limit > 0 {
|
||
if _, err := tx.ExecContext(ctx, `
|
||
INSERT INTO point_withdrawal_policy_counters (
|
||
app_code, source_asset_type, user_id, channel, period, period_key,
|
||
withdrawal_count, created_at_ms, updated_at_ms
|
||
) VALUES (?, ?, ?, ?, ?, ?, 0, ?, ?)
|
||
ON DUPLICATE KEY UPDATE period_key = VALUES(period_key)`,
|
||
appCode, sourceAssetType, userID, channel, limit.Period, limit.PeriodKey, nowMS, nowMS,
|
||
); err != nil {
|
||
return err
|
||
}
|
||
var current int64
|
||
if err := tx.QueryRowContext(ctx, `
|
||
SELECT withdrawal_count
|
||
FROM point_withdrawal_policy_counters
|
||
WHERE app_code = ? AND source_asset_type = ? AND user_id = ? AND channel = ? AND period = ? AND period_key = ?
|
||
FOR UPDATE`, appCode, sourceAssetType, userID, channel, limit.Period, limit.PeriodKey,
|
||
).Scan(¤t); err != nil {
|
||
return err
|
||
}
|
||
if current >= limit.Limit {
|
||
return xerr.NewWithMetadata(xerr.PointWithdrawalLimitReached, "point withdrawal limit reached", map[string]string{
|
||
"channel": channel,
|
||
"period": limit.Period,
|
||
"limit": strconv.FormatInt(limit.Limit, 10),
|
||
})
|
||
}
|
||
if _, err := tx.ExecContext(ctx, `
|
||
UPDATE point_withdrawal_policy_counters
|
||
SET withdrawal_count = withdrawal_count + 1, updated_at_ms = ?
|
||
WHERE app_code = ? AND source_asset_type = ? AND user_id = ? AND channel = ? AND period = ? AND period_key = ?`,
|
||
nowMS, appCode, sourceAssetType, userID, channel, limit.Period, limit.PeriodKey,
|
||
); err != nil {
|
||
return err
|
||
}
|
||
}
|
||
if channel != pointWithdrawalChannelPlatform {
|
||
// 转币商是单事务账变；缺政策时无需 reservation，直接保留旧工资余额的无限次转出行为。
|
||
return nil
|
||
}
|
||
// 即使 Host/Agency 没有政策也写 counted=false reservation：它不是次数限制，而是保证建单失败回滚后
|
||
// 原 freeze command 不能重放旧成功回执。policy_id/version=0 明确表示兼容旧工资余额而非命中政策。
|
||
_, err = tx.ExecContext(ctx, `
|
||
INSERT INTO point_withdrawal_limit_reservations (
|
||
app_code, source_asset_type, withdrawal_ref, freeze_command_id, user_id, channel,
|
||
policy_id, policy_version, period, period_key, counted, status, created_at_ms, released_at_ms
|
||
) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, 'active', ?, 0)`,
|
||
appCode, sourceAssetType, strings.TrimSpace(withdrawalRef), strings.TrimSpace(commandID), userID, channel,
|
||
limit.PolicyID, limit.PolicyVersion, limit.Period, limit.PeriodKey, found && limit.Limit > 0, nowMS,
|
||
)
|
||
if isMySQLDuplicateError(err) {
|
||
return xerr.New(xerr.IdempotencyConflict, "point withdrawal reservation already belongs to another command")
|
||
}
|
||
return err
|
||
}
|
||
|
||
func (r *Repository) resolveHostPolicyWithdrawalLimit(ctx context.Context, tx *sql.Tx, regionID int64, channel string, nowMS int64) (hostPolicyWithdrawalLimit, bool, error) {
|
||
return r.resolveHostPolicyWithdrawalLimitForPolicyType(ctx, tx, regionID, channel, nowMS, "")
|
||
}
|
||
|
||
func (r *Repository) resolveHostPolicyWithdrawalLimitForPolicyType(ctx context.Context, tx *sql.Tx, regionID int64, channel string, nowMS int64, policyType string) (hostPolicyWithdrawalLimit, bool, error) {
|
||
if nowMS <= 0 {
|
||
nowMS = time.Now().UTC().UnixMilli()
|
||
}
|
||
now := time.UnixMilli(nowMS).UTC()
|
||
cycleKey := now.Format("2006-01")
|
||
// 无政策兼容 reservation 仍需要稳定且可审计的周期字段；不计数，因此不会占用该自然月额度。
|
||
limit := hostPolicyWithdrawalLimit{Period: withdrawalLimitPeriodMonth, PeriodKey: cycleKey}
|
||
if regionID == 0 {
|
||
// 没有区域时不能猜测某个区域的 Host 政策；调用方必须显式决定失败关闭还是使用工资兼容语义。
|
||
return limit, false, nil
|
||
}
|
||
var policy ledger.HostSalaryPolicy
|
||
var found bool
|
||
var err error
|
||
if strings.EqualFold(strings.TrimSpace(policyType), ledger.HostPolicyTypePointDiamond) {
|
||
// 永久积分按照最新一份历史 POINT_DIAMOND 绑定继续提现；当前政策改回工资型不影响既有余额。
|
||
policy, err = r.resolvePointDiamondHostPolicy(ctx, tx, regionID, nowMS)
|
||
if err != nil {
|
||
if xerr.CodeOf(err) == xerr.PermissionDenied {
|
||
return limit, false, nil
|
||
}
|
||
return hostPolicyWithdrawalLimit{}, false, err
|
||
}
|
||
found = true
|
||
} else {
|
||
// 工资与历史 POINT 复用当前周期绑定，保留原有“仅紧邻上月复制”的限制语义。
|
||
policy, found, err = r.queryHostSalaryPolicyByCycle(ctx, tx, regionID, cycleKey, "", "")
|
||
if err != nil {
|
||
return hostPolicyWithdrawalLimit{}, false, err
|
||
}
|
||
if !found {
|
||
return limit, false, nil
|
||
}
|
||
}
|
||
resolved, resolveErr := hostPolicyWithdrawalLimitFromPolicy(policy, channel, now)
|
||
if resolveErr != nil {
|
||
return hostPolicyWithdrawalLimit{}, false, resolveErr
|
||
}
|
||
return resolved, true, nil
|
||
}
|
||
|
||
// hostPolicyWithdrawalLimitFromPolicy 让执行校验与 overview 复用同一份已发布政策快照和 UTC 周期算法。
|
||
func hostPolicyWithdrawalLimitFromPolicy(policy ledger.HostSalaryPolicy, channel string, now time.Time) (hostPolicyWithdrawalLimit, error) {
|
||
limit := hostPolicyWithdrawalLimit{
|
||
PolicyID: policy.PolicyID, PolicyVersion: policy.PolicyVersion,
|
||
Period: withdrawalLimitPeriodMonth, PeriodKey: now.UTC().Format("2006-01"),
|
||
}
|
||
switch channel {
|
||
case pointWithdrawalChannelCoinSeller:
|
||
limit.Period, limit.Limit = policy.CoinSellerWithdrawalLimitPeriod, policy.CoinSellerWithdrawalLimitCount
|
||
case pointWithdrawalChannelPlatform:
|
||
limit.Period, limit.Limit = policy.PlatformWithdrawalLimitPeriod, policy.PlatformWithdrawalLimitCount
|
||
limit.AllowedDays = strings.TrimSpace(policy.PlatformWithdrawalAllowedDays)
|
||
default:
|
||
return hostPolicyWithdrawalLimit{}, xerr.New(xerr.InvalidArgument, "host policy withdrawal channel is invalid")
|
||
}
|
||
limit.Period = strings.ToLower(strings.TrimSpace(limit.Period))
|
||
if limit.Limit < 0 {
|
||
return hostPolicyWithdrawalLimit{}, xerr.New(xerr.Internal, "host policy withdrawal limit is invalid")
|
||
}
|
||
if limit.Period == "" {
|
||
limit.Period = withdrawalLimitPeriodMonth
|
||
}
|
||
switch limit.Period {
|
||
case withdrawalLimitPeriodDay:
|
||
limit.PeriodKey = now.Format("2006-01-02")
|
||
case withdrawalLimitPeriodWeek:
|
||
// 周期固定为 UTC 周一 00:00 起；保存周一日期避免跨年 ISO 周编号歧义。
|
||
mondayOffset := (int(now.Weekday()) + 6) % 7
|
||
limit.PeriodKey = now.AddDate(0, 0, -mondayOffset).Format("2006-01-02")
|
||
case withdrawalLimitPeriodMonth:
|
||
limit.PeriodKey = now.UTC().Format("2006-01")
|
||
default:
|
||
return hostPolicyWithdrawalLimit{}, xerr.New(xerr.Internal, "host policy withdrawal period is invalid")
|
||
}
|
||
return limit, nil
|
||
}
|
||
|
||
// platformWithdrawalAllowedOnUTCDate 精确匹配 UTC 日历日；例如 30 号在二月不存在时不会自动改成月底。
|
||
// 空串兼容迁移前政策为每天可提现，异常快照则返回内部错误，不能静默放开限制。
|
||
func platformWithdrawalAllowedOnUTCDate(raw string, now time.Time) (bool, error) {
|
||
raw = strings.TrimSpace(raw)
|
||
if raw == "" {
|
||
return true, nil
|
||
}
|
||
currentDay := now.UTC().Day()
|
||
seen := make(map[int]struct{})
|
||
allowed := false
|
||
for _, part := range strings.Split(raw, ",") {
|
||
day, err := strconv.Atoi(strings.TrimSpace(part))
|
||
if err != nil || day < 1 || day > 31 {
|
||
return false, xerr.New(xerr.Internal, "point withdrawal allowed days policy is invalid")
|
||
}
|
||
if _, duplicate := seen[day]; duplicate {
|
||
return false, xerr.New(xerr.Internal, "point withdrawal allowed days policy is invalid")
|
||
}
|
||
seen[day] = struct{}{}
|
||
if day == currentDay {
|
||
allowed = true
|
||
}
|
||
}
|
||
return allowed, nil
|
||
}
|
||
|
||
// pointWithdrawalActionAvailability 只读命中 counter 主键，不加锁也不预占次数；实际动作仍会在
|
||
// 账变事务内锁行并重新校验，因此 overview 只是准确的当前快照，不承担授权事实。
|
||
func (r *Repository) pointWithdrawalActionAvailability(ctx context.Context, tx *sql.Tx, userID int64, sourceAssetType string, channel string, limit hostPolicyWithdrawalLimit, now time.Time) (ledger.PointWithdrawalActionAvailability, error) {
|
||
availability := ledger.PointWithdrawalActionAvailability{
|
||
Allowed: true, LimitPeriod: limit.Period, LimitCount: limit.Limit,
|
||
RemainingCount: -1, AllowedDays: limit.AllowedDays,
|
||
}
|
||
if channel == pointWithdrawalChannelPlatform {
|
||
allowed, err := platformWithdrawalAllowedOnUTCDate(limit.AllowedDays, now.UTC())
|
||
if err != nil {
|
||
return ledger.PointWithdrawalActionAvailability{}, err
|
||
}
|
||
if !allowed {
|
||
availability.Allowed = false
|
||
availability.BlockReason = "date_not_allowed"
|
||
}
|
||
}
|
||
if limit.Limit <= 0 {
|
||
return availability, nil
|
||
}
|
||
var used int64
|
||
err := tx.QueryRowContext(ctx, `
|
||
SELECT withdrawal_count
|
||
FROM point_withdrawal_policy_counters
|
||
WHERE app_code = ? AND source_asset_type = ? AND user_id = ?
|
||
AND channel = ? AND period = ? AND period_key = ?`,
|
||
appcode.FromContext(ctx), sourceAssetType, userID, channel, limit.Period, limit.PeriodKey,
|
||
).Scan(&used)
|
||
if errors.Is(err, sql.ErrNoRows) {
|
||
used = 0
|
||
} else if err != nil {
|
||
return ledger.PointWithdrawalActionAvailability{}, err
|
||
}
|
||
availability.UsedCount = used
|
||
availability.RemainingCount = limit.Limit - used
|
||
if availability.RemainingCount < 0 {
|
||
availability.RemainingCount = 0
|
||
}
|
||
if used >= limit.Limit {
|
||
availability.Allowed = false
|
||
availability.BlockReason = "limit_reached"
|
||
}
|
||
return availability, nil
|
||
}
|
||
|
||
// releaseHostPolicyWithdrawalLimitReservation 只服务 gateway 建单失败补偿。
|
||
// application_id 非空的人工驳回不会调用这里，因此驳回申请仍占用原限制周期的一次次数。
|
||
func (r *Repository) releaseHostPolicyWithdrawalLimitReservation(ctx context.Context, tx *sql.Tx, userID int64, sourceAssetType string, withdrawalRef string, rollbackCommandID string, nowMS int64) error {
|
||
withdrawalRef = strings.TrimSpace(withdrawalRef)
|
||
sourceAssetType = strings.ToUpper(strings.TrimSpace(sourceAssetType))
|
||
appCode := appcode.FromContext(ctx)
|
||
var reservedUserID int64
|
||
var channel string
|
||
var period string
|
||
var periodKey string
|
||
var counted bool
|
||
var status string
|
||
queryByRef := withdrawalRef != ""
|
||
var err error
|
||
if queryByRef {
|
||
err = tx.QueryRowContext(ctx, `
|
||
SELECT withdrawal_ref, user_id, channel, period, period_key, counted, status
|
||
FROM point_withdrawal_limit_reservations
|
||
WHERE app_code = ? AND source_asset_type = ? AND withdrawal_ref = ?
|
||
FOR UPDATE`, appCode, sourceAssetType, withdrawalRef,
|
||
).Scan(&withdrawalRef, &reservedUserID, &channel, &period, &periodKey, &counted, &status)
|
||
}
|
||
if !queryByRef || errors.Is(err, sql.ErrNoRows) {
|
||
// HTTP 重试会生成新的 request_id，但 freeze command_id 是客户端稳定幂等键；按 :rollback 约定回退到唯一命令索引。
|
||
freezeCommandID := strings.TrimSuffix(strings.TrimSpace(rollbackCommandID), ":rollback")
|
||
if freezeCommandID == strings.TrimSpace(rollbackCommandID) || freezeCommandID == "" {
|
||
if errors.Is(err, sql.ErrNoRows) || !queryByRef {
|
||
return nil
|
||
}
|
||
return err
|
||
}
|
||
err = tx.QueryRowContext(ctx, `
|
||
SELECT withdrawal_ref, user_id, channel, period, period_key, counted, status
|
||
FROM point_withdrawal_limit_reservations
|
||
WHERE app_code = ? AND source_asset_type = ? AND freeze_command_id = ?
|
||
FOR UPDATE`, appCode, sourceAssetType, freezeCommandID,
|
||
).Scan(&withdrawalRef, &reservedUserID, &channel, &period, &periodKey, &counted, &status)
|
||
}
|
||
if errors.Is(err, sql.ErrNoRows) {
|
||
// 旧冻结记录没有 reservation，释放余额仍按原兼容行为继续。
|
||
return nil
|
||
}
|
||
if err != nil {
|
||
return err
|
||
}
|
||
if reservedUserID != userID || channel != pointWithdrawalChannelPlatform {
|
||
return xerr.New(xerr.IdempotencyConflict, "withdrawal reservation does not match rollback")
|
||
}
|
||
if status == "released" {
|
||
return nil
|
||
}
|
||
if status != "active" {
|
||
return xerr.New(xerr.Internal, "withdrawal reservation status is invalid")
|
||
}
|
||
if counted {
|
||
// policy_id/version 只保留为命中政策审计；自然周期计数跨区域和同月显式政策切换连续，释放也必须命中共享周期键。
|
||
result, err := tx.ExecContext(ctx, `
|
||
UPDATE point_withdrawal_policy_counters
|
||
SET withdrawal_count = withdrawal_count - 1, updated_at_ms = ?
|
||
WHERE app_code = ? AND source_asset_type = ? AND user_id = ? AND channel = ? AND period = ? AND period_key = ?
|
||
AND withdrawal_count > 0`, nowMS, appCode, sourceAssetType, userID, channel, period, periodKey)
|
||
if err != nil {
|
||
return err
|
||
}
|
||
if affected, err := result.RowsAffected(); err != nil || affected != 1 {
|
||
if err != nil {
|
||
return err
|
||
}
|
||
return xerr.New(xerr.Internal, "withdrawal reservation counter is missing")
|
||
}
|
||
}
|
||
_, err = tx.ExecContext(ctx, `
|
||
UPDATE point_withdrawal_limit_reservations
|
||
SET status = 'released', released_at_ms = ?
|
||
WHERE app_code = ? AND source_asset_type = ? AND withdrawal_ref = ? AND status = 'active'`, nowMS, appCode, sourceAssetType, withdrawalRef)
|
||
return err
|
||
}
|
||
|
||
// hostPolicyWithdrawalReservationWasReleased 防止已补偿冻结的 command_id 重放旧成功回执。
|
||
// reservation 与冻结交易同事务落库，因此 active/不存在仍可按原幂等语义返回；released 必须要求客户端换新命令。
|
||
func (r *Repository) hostPolicyWithdrawalReservationWasReleased(ctx context.Context, tx *sql.Tx, sourceAssetType string, commandID string) (bool, error) {
|
||
var status string
|
||
err := tx.QueryRowContext(ctx, `
|
||
SELECT status
|
||
FROM point_withdrawal_limit_reservations
|
||
WHERE app_code = ? AND source_asset_type = ? AND freeze_command_id = ?`, appcode.FromContext(ctx), strings.ToUpper(strings.TrimSpace(sourceAssetType)), strings.TrimSpace(commandID)).Scan(&status)
|
||
if errors.Is(err, sql.ErrNoRows) {
|
||
return false, nil
|
||
}
|
||
if err != nil {
|
||
return false, err
|
||
}
|
||
return status == "released", nil
|
||
}
|