818 lines
31 KiB
Go
818 lines
31 KiB
Go
package mysql
|
||
|
||
import (
|
||
"context"
|
||
"crypto/sha1"
|
||
"database/sql"
|
||
"encoding/hex"
|
||
"encoding/json"
|
||
"errors"
|
||
"fmt"
|
||
"strings"
|
||
|
||
"hyapp/pkg/appcode"
|
||
"hyapp/pkg/idgen"
|
||
"hyapp/pkg/xerr"
|
||
lotteryengine "hyapp/services/activity-service/internal/domain/lotteryengine"
|
||
domain "hyapp/services/activity-service/internal/domain/wheel"
|
||
)
|
||
|
||
const wheelPPMScale int64 = 1_000_000
|
||
|
||
type wheelRTPWindow struct {
|
||
WheelID string
|
||
WindowIndex int64
|
||
TargetRTPPPM int64
|
||
ControlDraws int64
|
||
PaidDraws int64
|
||
WagerCoins int64
|
||
TargetPayout int64
|
||
ActualRTPValue int64
|
||
CarryPPM int64
|
||
Status string
|
||
}
|
||
|
||
type wheelPool struct {
|
||
WheelID string
|
||
Balance int64
|
||
ReserveFloor int64
|
||
TotalIn int64
|
||
TotalOut int64
|
||
}
|
||
|
||
type wheelCandidate struct {
|
||
Tier domain.Tier
|
||
}
|
||
|
||
// PublishWheelRuleConfig 新增转盘不可变配置版本；历史 draw record 永远引用旧版本，避免后台改配置后审计口径漂移。
|
||
func (r *Repository) PublishWheelRuleConfig(ctx context.Context, config domain.RuleConfig, nowMS int64) (domain.RuleConfig, error) {
|
||
if r == nil || r.db == nil {
|
||
return domain.RuleConfig{}, xerr.New(xerr.Unavailable, "mysql repository is not configured")
|
||
}
|
||
appCode := appcode.FromContext(ctx)
|
||
config.AppCode = appCode
|
||
config.WheelID = normalizeWheelID(config.WheelID)
|
||
config.Tiers = normalizeWheelTiers(config.Tiers)
|
||
tx, err := r.db.BeginTx(ctx, nil)
|
||
if err != nil {
|
||
return domain.RuleConfig{}, err
|
||
}
|
||
defer func() { _ = tx.Rollback() }()
|
||
var latest sql.NullInt64
|
||
if err := tx.QueryRowContext(ctx, `
|
||
SELECT MAX(rule_version)
|
||
FROM wheel_rule_versions
|
||
WHERE app_code = ? AND wheel_id = ?
|
||
FOR UPDATE`,
|
||
appCode, config.WheelID,
|
||
).Scan(&latest); err != nil {
|
||
return domain.RuleConfig{}, err
|
||
}
|
||
if latest.Valid {
|
||
config.RuleVersion = latest.Int64 + 1
|
||
} else {
|
||
config.RuleVersion = 1
|
||
}
|
||
if config.EffectiveFromMS <= 0 {
|
||
config.EffectiveFromMS = nowMS
|
||
}
|
||
config.CreatedAtMS = nowMS
|
||
if _, err := tx.ExecContext(ctx, `
|
||
INSERT INTO wheel_rule_versions (
|
||
app_code, wheel_id, rule_version, enabled, draw_price_coins, target_rtp_ppm, pool_rate_ppm,
|
||
settlement_window_draws, initial_pool_coins, pool_reserve_coins, max_single_rtp_payout,
|
||
effective_from_ms, created_by_admin_id, created_at_ms
|
||
) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`,
|
||
appCode, config.WheelID, config.RuleVersion, config.Enabled, config.DrawPriceCoins, config.TargetRTPPPM, config.PoolRatePPM,
|
||
config.SettlementWindowDraws, config.InitialPoolCoins, config.PoolReserveCoins, config.MaxSingleRTPPayout,
|
||
config.EffectiveFromMS, config.CreatedByAdminID, nowMS,
|
||
); err != nil {
|
||
return domain.RuleConfig{}, err
|
||
}
|
||
for _, tier := range config.Tiers {
|
||
if _, err := tx.ExecContext(ctx, `
|
||
INSERT INTO wheel_prize_tiers (
|
||
app_code, wheel_id, rule_version, tier_id, display_name, reward_type, reward_id, reward_count,
|
||
reward_coins, rtp_value_coins, weight_ppm, enabled, metadata_json
|
||
) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`,
|
||
appCode, config.WheelID, config.RuleVersion, tier.TierID, tier.DisplayName, tier.RewardType, tier.RewardID, tier.RewardCount,
|
||
tier.RewardCoins, tier.RTPValueCoins, tier.WeightPPM, tier.Enabled, wheelMetadataJSON(tier.MetadataJSON),
|
||
); err != nil {
|
||
return domain.RuleConfig{}, err
|
||
}
|
||
}
|
||
if err := tx.Commit(); err != nil {
|
||
return domain.RuleConfig{}, err
|
||
}
|
||
return config, nil
|
||
}
|
||
|
||
func (r *Repository) GetWheelRuleConfig(ctx context.Context, wheelID string) (domain.RuleConfig, bool, error) {
|
||
if r == nil || r.db == nil {
|
||
return domain.RuleConfig{}, false, xerr.New(xerr.Unavailable, "mysql repository is not configured")
|
||
}
|
||
return r.getWheelRuleConfig(ctx, r.db, appcode.FromContext(ctx), normalizeWheelID(wheelID), false)
|
||
}
|
||
|
||
// ExecuteWheelDraw 是转盘线上主事务：扣费事实已发生，事务内只做命中奖档、RTP 观察、奖池扣减和发放 outbox。
|
||
func (r *Repository) ExecuteWheelDraw(ctx context.Context, cmd domain.DrawCommand, nowMS int64) (domain.DrawResult, error) {
|
||
if r == nil || r.db == nil {
|
||
return domain.DrawResult{}, xerr.New(xerr.Unavailable, "mysql repository is not configured")
|
||
}
|
||
cmd.WheelID = normalizeWheelID(cmd.WheelID)
|
||
if cmd.DrawCount <= 0 {
|
||
cmd.DrawCount = 1
|
||
}
|
||
tx, err := r.db.BeginTx(ctx, nil)
|
||
if err != nil {
|
||
return domain.DrawResult{}, err
|
||
}
|
||
defer func() { _ = tx.Rollback() }()
|
||
appCode := appcode.FromContext(ctx)
|
||
existing, complete, err := r.collectWheelDrawsByCommand(ctx, tx, appCode, cmd.CommandID, cmd.DrawCount)
|
||
if err != nil {
|
||
return domain.DrawResult{}, err
|
||
}
|
||
if complete {
|
||
return aggregateWheelDrawResults(cmd, existing), nil
|
||
}
|
||
if len(existing) > 0 {
|
||
return domain.DrawResult{}, xerr.New(xerr.Conflict, "partial wheel draw command exists")
|
||
}
|
||
config, exists, err := r.getWheelRuleConfig(ctx, tx, appCode, cmd.WheelID, true)
|
||
if err != nil {
|
||
return domain.DrawResult{}, err
|
||
}
|
||
if !exists || !config.Enabled {
|
||
return domain.DrawResult{}, xerr.New(xerr.Conflict, "wheel is disabled")
|
||
}
|
||
if cmd.CoinSpent <= 0 || cmd.UserID <= 0 || strings.TrimSpace(cmd.CommandID) == "" || strings.TrimSpace(cmd.DeviceID) == "" {
|
||
return domain.DrawResult{}, xerr.New(xerr.InvalidArgument, "wheel draw command is incomplete")
|
||
}
|
||
expectedSpent := config.DrawPriceCoins * int64(cmd.DrawCount)
|
||
if cmd.CoinSpent != expectedSpent {
|
||
return domain.DrawResult{}, xerr.New(xerr.InvalidArgument, "wheel draw coin_spent does not match current price")
|
||
}
|
||
if cmd.PaidAtMS <= 0 {
|
||
cmd.PaidAtMS = nowMS
|
||
}
|
||
window, err := r.getOpenWheelRTPWindow(ctx, tx, appCode, config, nowMS)
|
||
if err != nil {
|
||
return domain.DrawResult{}, err
|
||
}
|
||
pool, err := r.getOrCreateWheelPool(ctx, tx, appCode, config, nowMS)
|
||
if err != nil {
|
||
return domain.DrawResult{}, err
|
||
}
|
||
results := make([]domain.DrawResult, 0, cmd.DrawCount)
|
||
var totalPoolIn, totalPoolOut int64
|
||
for i := int32(1); i <= cmd.DrawCount; i++ {
|
||
unitSpent := luckyDrawUnitSpend(cmd.CoinSpent, cmd.DrawCount, i)
|
||
totalPoolIn += unitSpent * config.PoolRatePPM / wheelPPMScale
|
||
pool.Balance += unitSpent * config.PoolRatePPM / wheelPPMScale
|
||
subCommand := wheelSubCommandID(cmd.CommandID, i, cmd.DrawCount)
|
||
result, poolOut, err := r.executeSingleWheelDraw(ctx, tx, appCode, cmd, subCommand, unitSpent, config, &window, &pool, nowMS)
|
||
if err != nil {
|
||
return domain.DrawResult{}, err
|
||
}
|
||
totalPoolOut += poolOut
|
||
results = append(results, result)
|
||
}
|
||
if err := r.persistWheelPoolDelta(ctx, tx, appCode, config.WheelID, totalPoolIn, totalPoolOut, nowMS); err != nil {
|
||
return domain.DrawResult{}, err
|
||
}
|
||
if err := r.updateWheelStats(ctx, tx, appCode, config.WheelID, cmd.UserID, cmd.CoinSpent, results, nowMS); err != nil {
|
||
return domain.DrawResult{}, err
|
||
}
|
||
if err := tx.Commit(); err != nil {
|
||
return domain.DrawResult{}, err
|
||
}
|
||
return aggregateWheelDrawResults(cmd, results), nil
|
||
}
|
||
|
||
func (r *Repository) ListWheelDraws(ctx context.Context, query domain.DrawQuery) ([]domain.DrawResult, int64, error) {
|
||
if r == nil || r.db == nil {
|
||
return nil, 0, xerr.New(xerr.Unavailable, "mysql repository is not configured")
|
||
}
|
||
whereSQL, args := wheelDrawWhereClause(appcode.FromContext(ctx), query)
|
||
var total int64
|
||
if err := r.db.QueryRowContext(ctx, `SELECT COUNT(*) FROM wheel_draw_records WHERE `+whereSQL, args...).Scan(&total); err != nil {
|
||
return nil, 0, err
|
||
}
|
||
page, pageSize := normalizeWheelPage(query.Page, query.PageSize)
|
||
args = append(args, int(pageSize), int((page-1)*pageSize))
|
||
rows, err := r.db.QueryContext(ctx, `
|
||
SELECT draw_id, command_id, wheel_id, rule_version, selected_tier_id, reward_type, reward_id,
|
||
reward_count, reward_coins, rtp_value_coins, reward_status, reward_transaction_id,
|
||
rtp_window_index, created_at_ms, COALESCE(CAST(prize_metadata_json AS CHAR), '{}')
|
||
FROM wheel_draw_records
|
||
WHERE `+whereSQL+`
|
||
ORDER BY created_at_ms DESC, draw_id DESC
|
||
LIMIT ? OFFSET ?`, args...)
|
||
if err != nil {
|
||
return nil, 0, err
|
||
}
|
||
defer rows.Close()
|
||
items := make([]domain.DrawResult, 0, pageSize)
|
||
for rows.Next() {
|
||
var item domain.DrawResult
|
||
if err := rows.Scan(&item.DrawID, &item.CommandID, &item.WheelID, &item.RuleVersion, &item.SelectedTierID, &item.RewardType, &item.RewardID,
|
||
&item.RewardCount, &item.RewardCoins, &item.RTPValueCoins, &item.RewardStatus, &item.WalletTransactionID,
|
||
&item.RTPWindowIndex, &item.CreatedAtMS, &item.MetadataJSON); err != nil {
|
||
return nil, 0, err
|
||
}
|
||
item.DrawIDs = []string{item.DrawID}
|
||
items = append(items, item)
|
||
}
|
||
return items, total, rows.Err()
|
||
}
|
||
|
||
func (r *Repository) GetWheelDrawSummary(ctx context.Context, query domain.DrawQuery) (domain.DrawSummary, error) {
|
||
if r == nil || r.db == nil {
|
||
return domain.DrawSummary{}, xerr.New(xerr.Unavailable, "mysql repository is not configured")
|
||
}
|
||
if query.UserID <= 0 && strings.TrimSpace(query.Status) == "" {
|
||
return r.getWheelDrawSummaryFromStats(ctx, query)
|
||
}
|
||
return r.getWheelDrawSummaryFromRecords(ctx, query)
|
||
}
|
||
|
||
func (r *Repository) getWheelDrawSummaryFromRecords(ctx context.Context, query domain.DrawQuery) (domain.DrawSummary, error) {
|
||
whereSQL, args := wheelDrawWhereClause(appcode.FromContext(ctx), query)
|
||
var summary domain.DrawSummary
|
||
summary.WheelID = normalizeWheelID(query.WheelID)
|
||
if err := r.db.QueryRowContext(ctx, `
|
||
SELECT
|
||
COUNT(*),
|
||
COUNT(DISTINCT user_id),
|
||
COALESCE(SUM(coin_spent), 0),
|
||
COALESCE(SUM(rtp_value_coins), 0),
|
||
COALESCE(SUM(CASE WHEN reward_status = 'pending' THEN 1 ELSE 0 END), 0),
|
||
COALESCE(SUM(CASE WHEN reward_status = 'granted' THEN 1 ELSE 0 END), 0),
|
||
COALESCE(SUM(CASE WHEN reward_status = 'failed' THEN 1 ELSE 0 END), 0)
|
||
FROM wheel_draw_records
|
||
WHERE `+whereSQL, args...).Scan(
|
||
&summary.TotalDraws,
|
||
&summary.UniqueUsers,
|
||
&summary.TotalSpentCoins,
|
||
&summary.TotalRTPValueCoins,
|
||
&summary.PendingDraws,
|
||
&summary.GrantedDraws,
|
||
&summary.FailedDraws,
|
||
); err != nil {
|
||
return domain.DrawSummary{}, err
|
||
}
|
||
summary.ActualRTPPPM = wheelRTPPPM(summary.TotalSpentCoins, summary.TotalRTPValueCoins)
|
||
return summary, nil
|
||
}
|
||
|
||
func (r *Repository) getWheelDrawSummaryFromStats(ctx context.Context, query domain.DrawQuery) (domain.DrawSummary, error) {
|
||
var summary domain.DrawSummary
|
||
summary.WheelID = normalizeWheelID(query.WheelID)
|
||
err := r.db.QueryRowContext(ctx, `
|
||
SELECT total_draws, unique_users, total_spent_coins, total_rtp_value_coins,
|
||
pending_draws, granted_draws, failed_draws
|
||
FROM wheel_draw_stats
|
||
WHERE app_code = ? AND wheel_id = ?`,
|
||
appcode.FromContext(ctx), summary.WheelID,
|
||
).Scan(
|
||
&summary.TotalDraws,
|
||
&summary.UniqueUsers,
|
||
&summary.TotalSpentCoins,
|
||
&summary.TotalRTPValueCoins,
|
||
&summary.PendingDraws,
|
||
&summary.GrantedDraws,
|
||
&summary.FailedDraws,
|
||
)
|
||
if errors.Is(err, sql.ErrNoRows) {
|
||
return summary, nil
|
||
}
|
||
if err != nil {
|
||
return domain.DrawSummary{}, err
|
||
}
|
||
summary.ActualRTPPPM = wheelRTPPPM(summary.TotalSpentCoins, summary.TotalRTPValueCoins)
|
||
return summary, nil
|
||
}
|
||
|
||
func (r *Repository) MarkWheelDrawsGranted(ctx context.Context, appCode string, drawIDs []string, transactionID string, nowMS int64) error {
|
||
if r == nil || r.db == nil {
|
||
return xerr.New(xerr.Unavailable, "mysql repository is not configured")
|
||
}
|
||
appCode = appcode.Normalize(appCode)
|
||
tx, err := r.db.BeginTx(ctx, nil)
|
||
if err != nil {
|
||
return err
|
||
}
|
||
defer func() { _ = tx.Rollback() }()
|
||
for _, drawID := range drawIDs {
|
||
drawID = strings.TrimSpace(drawID)
|
||
if drawID == "" {
|
||
continue
|
||
}
|
||
var wheelID, oldStatus string
|
||
if err := tx.QueryRowContext(ctx, `
|
||
SELECT wheel_id, reward_status
|
||
FROM wheel_draw_records
|
||
WHERE app_code = ? AND draw_id = ?
|
||
FOR UPDATE`,
|
||
appCode, drawID,
|
||
).Scan(&wheelID, &oldStatus); err != nil {
|
||
if errors.Is(err, sql.ErrNoRows) {
|
||
continue
|
||
}
|
||
return err
|
||
}
|
||
if oldStatus != domain.StatusGranted {
|
||
if _, err := tx.ExecContext(ctx, `
|
||
UPDATE wheel_draw_records
|
||
SET reward_status = 'granted', reward_transaction_id = ?, reward_failure_reason = '', updated_at_ms = ?
|
||
WHERE app_code = ? AND draw_id = ? AND reward_status <> 'granted'`,
|
||
transactionID, nowMS, appCode, drawID,
|
||
); err != nil {
|
||
return err
|
||
}
|
||
// 转盘统计是 draw 事务内即时累加的，后续发奖状态变化必须同步搬移计数，避免后台看到永久 pending。
|
||
if _, err := tx.ExecContext(ctx, `
|
||
UPDATE wheel_draw_stats
|
||
SET pending_draws = CASE WHEN pending_draws > 0 THEN pending_draws - 1 ELSE 0 END,
|
||
granted_draws = granted_draws + 1,
|
||
updated_at_ms = ?
|
||
WHERE app_code = ? AND wheel_id = ?`,
|
||
nowMS, appCode, wheelID,
|
||
); err != nil {
|
||
return err
|
||
}
|
||
}
|
||
if _, err := tx.ExecContext(ctx, `
|
||
UPDATE activity_outbox
|
||
SET status = 'delivered', locked_by = '', lock_until_ms = 0, last_error = '', updated_at_ms = ?
|
||
WHERE app_code = ? AND outbox_id = ?`,
|
||
nowMS, appCode, "wheel_reward_"+drawID,
|
||
); err != nil {
|
||
return err
|
||
}
|
||
}
|
||
return tx.Commit()
|
||
}
|
||
|
||
func (r *Repository) executeSingleWheelDraw(ctx context.Context, tx *sql.Tx, appCode string, cmd domain.DrawCommand, commandID string, unitSpent int64, config domain.RuleConfig, window *wheelRTPWindow, pool *wheelPool, nowMS int64) (domain.DrawResult, int64, error) {
|
||
candidates, limited := wheelPayableCandidates(config, pool)
|
||
if len(candidates) == 0 {
|
||
return domain.DrawResult{}, 0, xerr.New(xerr.Conflict, "wheel has no payable tier")
|
||
}
|
||
engineCandidates := make([]lotteryengine.Candidate, 0, len(candidates))
|
||
byID := make(map[string]domain.Tier, len(candidates))
|
||
for _, candidate := range candidates {
|
||
engineCandidates = append(engineCandidates, lotteryengine.Candidate{
|
||
ID: candidate.Tier.TierID,
|
||
Weight: candidate.Tier.WeightPPM,
|
||
PrizeKind: candidate.Tier.RewardType,
|
||
RTPValueCoins: candidate.Tier.RTPValueCoins,
|
||
})
|
||
byID[candidate.Tier.TierID] = candidate.Tier
|
||
}
|
||
selected, err := lotteryengine.SelectWeighted(engineCandidates)
|
||
if err != nil {
|
||
return domain.DrawResult{}, 0, err
|
||
}
|
||
tier := byID[selected.ID]
|
||
tier.RTPValueCoins = selected.RTPValueCoins
|
||
pool.Balance -= tier.RTPValueCoins
|
||
pool.TotalOut += tier.RTPValueCoins
|
||
window.PaidDraws++
|
||
window.WagerCoins += unitSpent
|
||
window.TargetPayout = window.WagerCoins * window.TargetRTPPPM / wheelPPMScale
|
||
window.ActualRTPValue += tier.RTPValueCoins
|
||
if _, err := tx.ExecContext(ctx, `
|
||
UPDATE wheel_rtp_windows
|
||
SET paid_draws = ?, wager_coins = ?, target_payout_coins = ?, actual_rtp_value_coins = ?,
|
||
carry_ppm = ?, updated_at_ms = ?
|
||
WHERE app_code = ? AND wheel_id = ? AND window_index = ?`,
|
||
window.PaidDraws, window.WagerCoins, window.TargetPayout, window.ActualRTPValue,
|
||
window.CarryPPM, nowMS, appCode, config.WheelID, window.WindowIndex,
|
||
); err != nil {
|
||
return domain.DrawResult{}, 0, err
|
||
}
|
||
drawID := idgen.New("wheel_draw")
|
||
status := domain.StatusGranted
|
||
if wheelDrawNeedsFulfillment(tier) {
|
||
status = domain.StatusPending
|
||
}
|
||
candidateJSON, _ := json.Marshal(map[string]any{"selected": tier.TierID, "limited": limited})
|
||
rtpJSON, _ := json.Marshal(map[string]any{"window_index": window.WindowIndex, "wager_coins": window.WagerCoins, "actual_rtp_value_coins": window.ActualRTPValue})
|
||
if _, err := tx.ExecContext(ctx, `
|
||
INSERT INTO wheel_draw_records (
|
||
app_code, draw_id, command_id, user_id, device_id, visible_region_id, wheel_id, coin_spent,
|
||
rule_version, rtp_window_index, selected_tier_id, reward_type, reward_id, reward_count,
|
||
reward_coins, rtp_value_coins, candidate_tiers_json, rtp_snapshot_json, prize_metadata_json,
|
||
reward_status, paid_at_ms, created_at_ms, updated_at_ms
|
||
) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`,
|
||
appCode, drawID, commandID, cmd.UserID, cmd.DeviceID, cmd.VisibleRegionID, config.WheelID, unitSpent,
|
||
config.RuleVersion, window.WindowIndex, tier.TierID, tier.RewardType, tier.RewardID, tier.RewardCount,
|
||
tier.RewardCoins, tier.RTPValueCoins, string(candidateJSON), string(rtpJSON), wheelMetadataJSON(tier.MetadataJSON),
|
||
status, cmd.PaidAtMS, nowMS, nowMS,
|
||
); err != nil {
|
||
return domain.DrawResult{}, 0, err
|
||
}
|
||
if status == domain.StatusPending {
|
||
if err := r.insertWheelRewardOutbox(ctx, tx, appCode, drawID, commandID, cmd, tier, nowMS); err != nil {
|
||
return domain.DrawResult{}, 0, err
|
||
}
|
||
}
|
||
return domain.DrawResult{
|
||
DrawID: drawID,
|
||
CommandID: commandID,
|
||
WheelID: config.WheelID,
|
||
RuleVersion: config.RuleVersion,
|
||
SelectedTierID: tier.TierID,
|
||
RewardType: tier.RewardType,
|
||
RewardID: tier.RewardID,
|
||
RewardCount: tier.RewardCount,
|
||
RewardCoins: tier.RewardCoins,
|
||
RTPValueCoins: tier.RTPValueCoins,
|
||
RewardStatus: status,
|
||
RTPWindowIndex: window.WindowIndex,
|
||
ActualRTPPPM: wheelRTPPPM(window.WagerCoins, window.ActualRTPValue),
|
||
CreatedAtMS: nowMS,
|
||
MetadataJSON: wheelMetadataJSON(tier.MetadataJSON),
|
||
}, tier.RTPValueCoins, nil
|
||
}
|
||
|
||
func (r *Repository) getWheelRuleConfig(ctx context.Context, queryer interface {
|
||
QueryRowContext(context.Context, string, ...any) *sql.Row
|
||
QueryContext(context.Context, string, ...any) (*sql.Rows, error)
|
||
}, appCode, wheelID string, forUpdate bool) (domain.RuleConfig, bool, error) {
|
||
lockSQL := ""
|
||
if forUpdate {
|
||
lockSQL = " FOR UPDATE"
|
||
}
|
||
row := queryer.QueryRowContext(ctx, `
|
||
SELECT app_code, wheel_id, rule_version, enabled, draw_price_coins, target_rtp_ppm, pool_rate_ppm,
|
||
settlement_window_draws, initial_pool_coins, pool_reserve_coins, max_single_rtp_payout,
|
||
effective_from_ms, created_by_admin_id, created_at_ms
|
||
FROM wheel_rule_versions
|
||
WHERE app_code = ? AND wheel_id = ? AND enabled = 1
|
||
ORDER BY rule_version DESC
|
||
LIMIT 1`+lockSQL,
|
||
appCode, wheelID,
|
||
)
|
||
var config domain.RuleConfig
|
||
if err := row.Scan(&config.AppCode, &config.WheelID, &config.RuleVersion, &config.Enabled, &config.DrawPriceCoins, &config.TargetRTPPPM, &config.PoolRatePPM,
|
||
&config.SettlementWindowDraws, &config.InitialPoolCoins, &config.PoolReserveCoins, &config.MaxSingleRTPPayout,
|
||
&config.EffectiveFromMS, &config.CreatedByAdminID, &config.CreatedAtMS); err != nil {
|
||
if errors.Is(err, sql.ErrNoRows) {
|
||
return domain.RuleConfig{}, false, nil
|
||
}
|
||
return domain.RuleConfig{}, false, err
|
||
}
|
||
rows, err := queryer.QueryContext(ctx, `
|
||
SELECT tier_id, display_name, reward_type, reward_id, reward_count, reward_coins, rtp_value_coins, weight_ppm, enabled,
|
||
COALESCE(CAST(metadata_json AS CHAR), '{}')
|
||
FROM wheel_prize_tiers
|
||
WHERE app_code = ? AND wheel_id = ? AND rule_version = ?
|
||
ORDER BY tier_id`,
|
||
appCode, config.WheelID, config.RuleVersion,
|
||
)
|
||
if err != nil {
|
||
return domain.RuleConfig{}, false, err
|
||
}
|
||
defer rows.Close()
|
||
for rows.Next() {
|
||
var tier domain.Tier
|
||
if err := rows.Scan(&tier.TierID, &tier.DisplayName, &tier.RewardType, &tier.RewardID, &tier.RewardCount, &tier.RewardCoins, &tier.RTPValueCoins, &tier.WeightPPM, &tier.Enabled, &tier.MetadataJSON); err != nil {
|
||
return domain.RuleConfig{}, false, err
|
||
}
|
||
config.Tiers = append(config.Tiers, tier)
|
||
}
|
||
if err := rows.Err(); err != nil {
|
||
return domain.RuleConfig{}, false, err
|
||
}
|
||
return config, true, nil
|
||
}
|
||
|
||
func (r *Repository) getOpenWheelRTPWindow(ctx context.Context, tx *sql.Tx, appCode string, config domain.RuleConfig, nowMS int64) (wheelRTPWindow, error) {
|
||
window, exists, err := r.getLatestWheelRTPWindow(ctx, tx, appCode, config.WheelID, true)
|
||
if err != nil {
|
||
return wheelRTPWindow{}, err
|
||
}
|
||
if !exists {
|
||
return r.createWheelRTPWindow(ctx, tx, appCode, config.WheelID, 1, 0, config.SettlementWindowDraws, config.TargetRTPPPM, nowMS)
|
||
}
|
||
if window.Status == "open" && window.PaidDraws < window.ControlDraws {
|
||
return window, nil
|
||
}
|
||
if window.Status == "open" {
|
||
if _, err := tx.ExecContext(ctx, `UPDATE wheel_rtp_windows SET status = 'closed', updated_at_ms = ? WHERE app_code = ? AND wheel_id = ? AND window_index = ?`, nowMS, appCode, config.WheelID, window.WindowIndex); err != nil {
|
||
return wheelRTPWindow{}, err
|
||
}
|
||
}
|
||
return r.createWheelRTPWindow(ctx, tx, appCode, config.WheelID, window.WindowIndex+1, window.CarryPPM, config.SettlementWindowDraws, config.TargetRTPPPM, nowMS)
|
||
}
|
||
|
||
func (r *Repository) getLatestWheelRTPWindow(ctx context.Context, tx *sql.Tx, appCode, wheelID string, forUpdate bool) (wheelRTPWindow, bool, error) {
|
||
lockSQL := ""
|
||
if forUpdate {
|
||
lockSQL = " FOR UPDATE"
|
||
}
|
||
row := tx.QueryRowContext(ctx, `
|
||
SELECT wheel_id, window_index, target_rtp_ppm, control_window_draws, paid_draws, wager_coins,
|
||
target_payout_coins, actual_rtp_value_coins, carry_ppm, status
|
||
FROM wheel_rtp_windows
|
||
WHERE app_code = ? AND wheel_id = ?
|
||
ORDER BY window_index DESC LIMIT 1`+lockSQL,
|
||
appCode, wheelID,
|
||
)
|
||
var window wheelRTPWindow
|
||
if err := row.Scan(&window.WheelID, &window.WindowIndex, &window.TargetRTPPPM, &window.ControlDraws, &window.PaidDraws, &window.WagerCoins,
|
||
&window.TargetPayout, &window.ActualRTPValue, &window.CarryPPM, &window.Status); err != nil {
|
||
if errors.Is(err, sql.ErrNoRows) {
|
||
return wheelRTPWindow{}, false, nil
|
||
}
|
||
return wheelRTPWindow{}, false, err
|
||
}
|
||
return window, true, nil
|
||
}
|
||
|
||
func (r *Repository) createWheelRTPWindow(ctx context.Context, tx *sql.Tx, appCode, wheelID string, index, carry, windowDraws, targetPPM, nowMS int64) (wheelRTPWindow, error) {
|
||
if windowDraws <= 0 {
|
||
windowDraws = 1
|
||
}
|
||
window := wheelRTPWindow{WheelID: wheelID, WindowIndex: index, TargetRTPPPM: targetPPM, ControlDraws: windowDraws, CarryPPM: carry, Status: "open"}
|
||
if _, err := tx.ExecContext(ctx, `
|
||
INSERT INTO wheel_rtp_windows (
|
||
app_code, wheel_id, window_index, target_rtp_ppm, control_window_draws,
|
||
paid_draws, wager_coins, target_payout_coins, actual_rtp_value_coins, carry_ppm, status, created_at_ms, updated_at_ms
|
||
) VALUES (?, ?, ?, ?, ?, 0, 0, 0, 0, ?, 'open', ?, ?)`,
|
||
appCode, wheelID, index, targetPPM, windowDraws, carry, nowMS, nowMS,
|
||
); err != nil {
|
||
return wheelRTPWindow{}, err
|
||
}
|
||
return window, nil
|
||
}
|
||
|
||
func (r *Repository) getOrCreateWheelPool(ctx context.Context, tx *sql.Tx, appCode string, config domain.RuleConfig, nowMS int64) (wheelPool, error) {
|
||
row := tx.QueryRowContext(ctx, `SELECT wheel_id, balance, reserve_floor, total_in, total_out FROM wheel_pools WHERE app_code = ? AND wheel_id = ? FOR UPDATE`, appCode, config.WheelID)
|
||
var pool wheelPool
|
||
if err := row.Scan(&pool.WheelID, &pool.Balance, &pool.ReserveFloor, &pool.TotalIn, &pool.TotalOut); err != nil {
|
||
if !errors.Is(err, sql.ErrNoRows) {
|
||
return wheelPool{}, err
|
||
}
|
||
pool = wheelPool{WheelID: config.WheelID, Balance: config.InitialPoolCoins, ReserveFloor: config.PoolReserveCoins}
|
||
if _, err := tx.ExecContext(ctx, `INSERT INTO wheel_pools (app_code, wheel_id, balance, reserve_floor, total_in, total_out, created_at_ms, updated_at_ms) VALUES (?, ?, ?, ?, 0, 0, ?, ?)`,
|
||
appCode, pool.WheelID, pool.Balance, pool.ReserveFloor, nowMS, nowMS); err != nil {
|
||
return wheelPool{}, err
|
||
}
|
||
}
|
||
return pool, nil
|
||
}
|
||
|
||
func (r *Repository) persistWheelPoolDelta(ctx context.Context, tx *sql.Tx, appCode, wheelID string, in, out, nowMS int64) error {
|
||
_, err := tx.ExecContext(ctx, `
|
||
UPDATE wheel_pools
|
||
SET balance = balance + ? - ?, total_in = total_in + ?, total_out = total_out + ?, updated_at_ms = ?
|
||
WHERE app_code = ? AND wheel_id = ?`,
|
||
in, out, in, out, nowMS, appCode, wheelID,
|
||
)
|
||
return err
|
||
}
|
||
|
||
func wheelPayableCandidates(config domain.RuleConfig, pool *wheelPool) ([]wheelCandidate, map[string]bool) {
|
||
limited := map[string]bool{}
|
||
capacity := pool.Balance - pool.ReserveFloor
|
||
if capacity < 0 {
|
||
capacity = 0
|
||
}
|
||
if config.MaxSingleRTPPayout > 0 && config.MaxSingleRTPPayout < capacity {
|
||
capacity = config.MaxSingleRTPPayout
|
||
}
|
||
candidates := make([]wheelCandidate, 0, len(config.Tiers))
|
||
for _, tier := range config.Tiers {
|
||
if !tier.Enabled {
|
||
continue
|
||
}
|
||
tier.RewardType = lotteryengine.NormalizePrizeKind(tier.RewardType)
|
||
tier.RTPValueCoins = lotteryengine.NormalizeRTPValueCoins(tier.RewardType, tier.RTPValueCoins)
|
||
if tier.RTPValueCoins > capacity {
|
||
limited["pool_cap"] = true
|
||
continue
|
||
}
|
||
candidates = append(candidates, wheelCandidate{Tier: tier})
|
||
}
|
||
return candidates, limited
|
||
}
|
||
|
||
func (r *Repository) updateWheelStats(ctx context.Context, tx *sql.Tx, appCode, wheelID string, userID int64, totalSpent int64, results []domain.DrawResult, nowMS int64) error {
|
||
var spent, rtpValue, pending, granted, failed int64
|
||
spent = totalSpent
|
||
for _, result := range results {
|
||
rtpValue += result.RTPValueCoins
|
||
switch result.RewardStatus {
|
||
case domain.StatusPending:
|
||
pending++
|
||
case domain.StatusFailed:
|
||
failed++
|
||
default:
|
||
granted++
|
||
}
|
||
}
|
||
if _, err := tx.ExecContext(ctx, `INSERT IGNORE INTO wheel_draw_stat_users (app_code, wheel_id, user_id, created_at_ms) VALUES (?, ?, ?, ?)`, appCode, wheelID, userID, nowMS); err != nil {
|
||
return err
|
||
}
|
||
var uniqueDelta int64
|
||
if err := tx.QueryRowContext(ctx, `SELECT COUNT(*) FROM wheel_draw_stat_users WHERE app_code = ? AND wheel_id = ?`, appCode, wheelID).Scan(&uniqueDelta); err != nil {
|
||
return err
|
||
}
|
||
if _, err := tx.ExecContext(ctx, `
|
||
INSERT INTO wheel_draw_stats (
|
||
app_code, wheel_id, total_draws, unique_users, total_spent_coins, total_rtp_value_coins,
|
||
pending_draws, granted_draws, failed_draws, created_at_ms, updated_at_ms
|
||
) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
|
||
ON DUPLICATE KEY UPDATE
|
||
total_draws = total_draws + VALUES(total_draws),
|
||
unique_users = ?,
|
||
total_spent_coins = total_spent_coins + VALUES(total_spent_coins),
|
||
total_rtp_value_coins = total_rtp_value_coins + VALUES(total_rtp_value_coins),
|
||
pending_draws = pending_draws + VALUES(pending_draws),
|
||
granted_draws = granted_draws + VALUES(granted_draws),
|
||
failed_draws = failed_draws + VALUES(failed_draws),
|
||
updated_at_ms = VALUES(updated_at_ms)`,
|
||
appCode, wheelID, int64(len(results)), uniqueDelta, spent, rtpValue, pending, granted, failed, nowMS, nowMS, uniqueDelta,
|
||
); err != nil {
|
||
return err
|
||
}
|
||
return nil
|
||
}
|
||
|
||
func (r *Repository) insertWheelRewardOutbox(ctx context.Context, tx *sql.Tx, appCode, drawID, commandID string, cmd domain.DrawCommand, tier domain.Tier, nowMS int64) error {
|
||
payload, _ := json.Marshal(map[string]any{
|
||
"app_code": appCode,
|
||
"draw_id": drawID,
|
||
"command_id": commandID,
|
||
"user_id": cmd.UserID,
|
||
"wheel_id": cmd.WheelID,
|
||
"selected_tier_id": tier.TierID,
|
||
"reward_type": tier.RewardType,
|
||
"reward_id": tier.RewardID,
|
||
"reward_count": tier.RewardCount,
|
||
"reward_coins": tier.RewardCoins,
|
||
"visible_region_id": cmd.VisibleRegionID,
|
||
"created_at_ms": nowMS,
|
||
})
|
||
return r.insertLuckyDrawOutbox(ctx, tx, appCode, "wheel_reward_"+drawID, domain.EventTypeWheelRewardSettlement, payload, nowMS)
|
||
}
|
||
|
||
func (r *Repository) collectWheelDrawsByCommand(ctx context.Context, tx *sql.Tx, appCode, commandID string, drawCount int32) ([]domain.DrawResult, bool, error) {
|
||
if drawCount <= 0 {
|
||
drawCount = 1
|
||
}
|
||
results := make([]domain.DrawResult, 0, drawCount)
|
||
for i := int32(1); i <= drawCount; i++ {
|
||
sub := wheelSubCommandID(commandID, i, drawCount)
|
||
row := tx.QueryRowContext(ctx, `
|
||
SELECT draw_id, command_id, wheel_id, rule_version, selected_tier_id, reward_type, reward_id, reward_count,
|
||
reward_coins, rtp_value_coins, reward_status, reward_transaction_id, rtp_window_index, created_at_ms,
|
||
COALESCE(CAST(prize_metadata_json AS CHAR), '{}')
|
||
FROM wheel_draw_records
|
||
WHERE app_code = ? AND command_id = ?`,
|
||
appCode, sub,
|
||
)
|
||
var result domain.DrawResult
|
||
if err := row.Scan(&result.DrawID, &result.CommandID, &result.WheelID, &result.RuleVersion, &result.SelectedTierID, &result.RewardType, &result.RewardID, &result.RewardCount,
|
||
&result.RewardCoins, &result.RTPValueCoins, &result.RewardStatus, &result.WalletTransactionID, &result.RTPWindowIndex, &result.CreatedAtMS, &result.MetadataJSON); err != nil {
|
||
if errors.Is(err, sql.ErrNoRows) {
|
||
continue
|
||
}
|
||
return nil, false, err
|
||
}
|
||
results = append(results, result)
|
||
}
|
||
return results, int32(len(results)) == drawCount, nil
|
||
}
|
||
|
||
func aggregateWheelDrawResults(cmd domain.DrawCommand, results []domain.DrawResult) domain.DrawResult {
|
||
if len(results) == 0 {
|
||
return domain.DrawResult{}
|
||
}
|
||
aggregate := results[0]
|
||
aggregate.CommandID = cmd.CommandID
|
||
aggregate.DrawIDs = make([]string, 0, len(results))
|
||
if len(results) == 1 {
|
||
aggregate.DrawIDs = []string{aggregate.DrawID}
|
||
return aggregate
|
||
}
|
||
aggregate.SelectedTierID = "batch"
|
||
aggregate.RewardType = "batch"
|
||
aggregate.RewardID = ""
|
||
aggregate.RewardCount = 0
|
||
aggregate.RewardCoins = 0
|
||
aggregate.RTPValueCoins = 0
|
||
aggregate.RewardStatus = domain.StatusGranted
|
||
for _, result := range results {
|
||
aggregate.DrawIDs = append(aggregate.DrawIDs, result.DrawID)
|
||
aggregate.RewardCoins += result.RewardCoins
|
||
aggregate.RTPValueCoins += result.RTPValueCoins
|
||
if result.RewardStatus == domain.StatusPending {
|
||
aggregate.RewardStatus = domain.StatusPending
|
||
} else if aggregate.RewardStatus != domain.StatusPending && result.RewardStatus == domain.StatusFailed {
|
||
aggregate.RewardStatus = domain.StatusFailed
|
||
}
|
||
}
|
||
return aggregate
|
||
}
|
||
|
||
func wheelSubCommandID(commandID string, drawIndex int32, drawCount int32) string {
|
||
if drawCount <= 1 {
|
||
return strings.TrimSpace(commandID)
|
||
}
|
||
suffix := fmt.Sprintf("#%06d", drawIndex)
|
||
const maxCommandIDLength = 128
|
||
commandID = strings.TrimSpace(commandID)
|
||
if len(commandID)+len(suffix) <= maxCommandIDLength {
|
||
return commandID + suffix
|
||
}
|
||
sum := sha1.Sum([]byte(commandID))
|
||
hash := hex.EncodeToString(sum[:])[:10]
|
||
hashSuffix := "#" + hash + suffix
|
||
headLen := maxCommandIDLength - len(hashSuffix)
|
||
if headLen < 0 {
|
||
headLen = 0
|
||
}
|
||
if len(commandID) > headLen {
|
||
commandID = commandID[:headLen]
|
||
}
|
||
return commandID + hashSuffix
|
||
}
|
||
|
||
func normalizeWheelID(wheelID string) string {
|
||
wheelID = strings.TrimSpace(wheelID)
|
||
if wheelID == "" {
|
||
return "default"
|
||
}
|
||
return wheelID
|
||
}
|
||
|
||
func normalizeWheelTiers(tiers []domain.Tier) []domain.Tier {
|
||
out := make([]domain.Tier, 0, len(tiers))
|
||
for _, tier := range tiers {
|
||
tier.TierID = strings.TrimSpace(tier.TierID)
|
||
tier.RewardType = lotteryengine.NormalizePrizeKind(tier.RewardType)
|
||
tier.RTPValueCoins = lotteryengine.NormalizeRTPValueCoins(tier.RewardType, tier.RTPValueCoins)
|
||
tier.MetadataJSON = wheelMetadataJSON(tier.MetadataJSON)
|
||
if tier.RewardCount <= 0 {
|
||
tier.RewardCount = 1
|
||
}
|
||
out = append(out, tier)
|
||
}
|
||
return out
|
||
}
|
||
|
||
func wheelDrawNeedsFulfillment(tier domain.Tier) bool {
|
||
return tier.RewardCoins > 0 || strings.TrimSpace(tier.RewardID) != ""
|
||
}
|
||
|
||
func wheelMetadataJSON(raw string) string {
|
||
raw = strings.TrimSpace(raw)
|
||
if raw == "" || !json.Valid([]byte(raw)) {
|
||
return "{}"
|
||
}
|
||
return raw
|
||
}
|
||
|
||
func wheelRTPPPM(wager, payout int64) int64 {
|
||
if wager <= 0 {
|
||
return 0
|
||
}
|
||
return payout * wheelPPMScale / wager
|
||
}
|
||
|
||
func wheelDrawWhereClause(appCode string, query domain.DrawQuery) (string, []any) {
|
||
where := []string{"app_code = ?"}
|
||
args := []any{appCode}
|
||
if query.WheelID != "" {
|
||
where = append(where, "wheel_id = ?")
|
||
args = append(args, normalizeWheelID(query.WheelID))
|
||
}
|
||
if query.UserID > 0 {
|
||
where = append(where, "user_id = ?")
|
||
args = append(args, query.UserID)
|
||
}
|
||
if query.Status != "" {
|
||
where = append(where, "reward_status = ?")
|
||
args = append(args, strings.TrimSpace(query.Status))
|
||
}
|
||
return strings.Join(where, " AND "), args
|
||
}
|
||
|
||
func normalizeWheelPage(page int32, pageSize int32) (int32, int32) {
|
||
if page <= 0 {
|
||
page = 1
|
||
}
|
||
if pageSize <= 0 {
|
||
pageSize = 20
|
||
}
|
||
if pageSize > 100 {
|
||
pageSize = 100
|
||
}
|
||
return page, pageSize
|
||
}
|