394 lines
17 KiB
Go
394 lines
17 KiB
Go
package mysql
|
||
|
||
import (
|
||
"context"
|
||
"database/sql"
|
||
"errors"
|
||
"fmt"
|
||
"strings"
|
||
|
||
_ "github.com/go-sql-driver/mysql"
|
||
"hyapp/pkg/appcode"
|
||
"hyapp/pkg/xerr"
|
||
activitydomain "hyapp/services/activity-service/internal/domain/activity"
|
||
)
|
||
|
||
// Repository 是 activity-service 的 MySQL 存储入口。
|
||
type Repository struct {
|
||
db *sql.DB
|
||
}
|
||
|
||
// Open 创建 MySQL 连接池并做启动 ping。
|
||
func Open(ctx context.Context, dsn string) (*Repository, error) {
|
||
if strings.TrimSpace(dsn) == "" {
|
||
return nil, xerr.New(xerr.InvalidArgument, "mysql_dsn is required")
|
||
}
|
||
|
||
db, err := sql.Open("mysql", dsn)
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
|
||
if err := db.PingContext(ctx); err != nil {
|
||
_ = db.Close()
|
||
return nil, err
|
||
}
|
||
|
||
return &Repository{db: db}, nil
|
||
}
|
||
|
||
// Migrate performs narrow, idempotent local schema fixes for activity-service.
|
||
func (r *Repository) Migrate(ctx context.Context) error {
|
||
if r == nil || r.db == nil {
|
||
return xerr.New(xerr.Unavailable, "mysql repository is not configured")
|
||
}
|
||
if err := r.ensureLuckyDrawPoolID(ctx); err != nil {
|
||
return err
|
||
}
|
||
if err := r.ensureLuckyGiftOutboxColumns(ctx); err != nil {
|
||
return err
|
||
}
|
||
if err := r.ensureLuckyDrawRewardSettlementColumns(ctx); err != nil {
|
||
return err
|
||
}
|
||
if err := r.ensureLuckyGiftRuleVersionTables(ctx); err != nil {
|
||
return err
|
||
}
|
||
if err := r.ensureLuckyUserStateFlowColumns(ctx); err != nil {
|
||
return err
|
||
}
|
||
if err := r.ensureDefaultLuckyGiftRules(ctx); err != nil {
|
||
return err
|
||
}
|
||
return nil
|
||
}
|
||
|
||
func (r *Repository) ensureLuckyUserStateFlowColumns(ctx context.Context) error {
|
||
// 用户阶段改按金币流水折算等价抽数，状态表必须持久保存累计流水和连续未中奖次数，不能再只依赖 paid_draws。
|
||
additions := []struct {
|
||
name string
|
||
sql string
|
||
}{
|
||
{"cumulative_wager_coins", `ALTER TABLE lucky_user_states ADD COLUMN cumulative_wager_coins BIGINT NOT NULL DEFAULT 0 COMMENT '该用户在当前奖池累计消耗金币' AFTER paid_draws`},
|
||
{"equivalent_draws", `ALTER TABLE lucky_user_states ADD COLUMN equivalent_draws BIGINT NOT NULL DEFAULT 0 COMMENT '累计金币按规则参考价格折算后的等价抽数' AFTER cumulative_wager_coins`},
|
||
{"loss_streak", `ALTER TABLE lucky_user_states ADD COLUMN loss_streak BIGINT NOT NULL DEFAULT 0 COMMENT '连续未获得可见奖励次数，中奖后清零' AFTER equivalent_draws`},
|
||
}
|
||
for _, addition := range additions {
|
||
exists, err := r.columnExists(ctx, "lucky_user_states", addition.name)
|
||
if err != nil {
|
||
return err
|
||
}
|
||
if !exists {
|
||
if _, err := r.db.ExecContext(ctx, addition.sql); err != nil {
|
||
return err
|
||
}
|
||
}
|
||
}
|
||
return nil
|
||
}
|
||
|
||
// Close 释放 MySQL 连接池。
|
||
func (r *Repository) Close() error {
|
||
if r == nil || r.db == nil {
|
||
return nil
|
||
}
|
||
|
||
return r.db.Close()
|
||
}
|
||
|
||
// Ping 验证 MySQL 当前可用。
|
||
func (r *Repository) Ping(ctx context.Context) error {
|
||
if r == nil || r.db == nil {
|
||
return xerr.New(xerr.Unavailable, "mysql repository is not configured")
|
||
}
|
||
|
||
return r.db.PingContext(ctx)
|
||
}
|
||
|
||
func (r *Repository) ensureLuckyDrawPoolID(ctx context.Context) error {
|
||
const table = "lucky_draw_records"
|
||
hasPoolID, err := r.columnExists(ctx, table, "pool_id")
|
||
if err != nil {
|
||
return err
|
||
}
|
||
if !hasPoolID {
|
||
if _, err := r.db.ExecContext(ctx, `
|
||
ALTER TABLE lucky_draw_records
|
||
ADD COLUMN pool_id VARCHAR(96) NOT NULL DEFAULT '' COMMENT '幸运礼物奖池 ID' AFTER anchor_id`); err != nil {
|
||
return err
|
||
}
|
||
}
|
||
if _, err := r.db.ExecContext(ctx, `UPDATE lucky_draw_records SET pool_id = gift_id WHERE pool_id = ''`); err != nil {
|
||
return err
|
||
}
|
||
hasPoolIndex, err := r.indexExists(ctx, table, "idx_lucky_draw_pool")
|
||
if err != nil {
|
||
return err
|
||
}
|
||
if !hasPoolIndex {
|
||
if _, err := r.db.ExecContext(ctx, `CREATE INDEX idx_lucky_draw_pool ON lucky_draw_records (app_code, pool_id, created_at_ms)`); err != nil {
|
||
return err
|
||
}
|
||
}
|
||
return nil
|
||
}
|
||
|
||
func (r *Repository) ensureLuckyGiftOutboxColumns(ctx context.Context) error {
|
||
additions := []struct {
|
||
name string
|
||
sql string
|
||
}{
|
||
{"locked_by", `ALTER TABLE activity_outbox ADD COLUMN locked_by VARCHAR(128) NOT NULL DEFAULT '' COMMENT '当前锁定 worker' AFTER next_retry_at_ms`},
|
||
{"lock_until_ms", `ALTER TABLE activity_outbox ADD COLUMN lock_until_ms BIGINT NOT NULL DEFAULT 0 COMMENT '锁过期时间，UTC epoch ms' AFTER locked_by`},
|
||
{"last_error", `ALTER TABLE activity_outbox ADD COLUMN last_error VARCHAR(512) NOT NULL DEFAULT '' COMMENT '最后一次失败原因' AFTER lock_until_ms`},
|
||
}
|
||
for _, addition := range additions {
|
||
exists, err := r.columnExists(ctx, "activity_outbox", addition.name)
|
||
if err != nil {
|
||
return err
|
||
}
|
||
if !exists {
|
||
if _, err := r.db.ExecContext(ctx, addition.sql); err != nil {
|
||
return err
|
||
}
|
||
}
|
||
}
|
||
return nil
|
||
}
|
||
|
||
func (r *Repository) ensureLuckyDrawRewardSettlementColumns(ctx context.Context) error {
|
||
additions := []struct {
|
||
name string
|
||
sql string
|
||
}{
|
||
{"reward_transaction_id", `ALTER TABLE lucky_draw_records ADD COLUMN reward_transaction_id VARCHAR(128) NOT NULL DEFAULT '' COMMENT '返奖钱包交易 ID' AFTER reward_status`},
|
||
{"reward_failure_reason", `ALTER TABLE lucky_draw_records ADD COLUMN reward_failure_reason VARCHAR(255) NOT NULL DEFAULT '' COMMENT '返奖失败原因' AFTER reward_transaction_id`},
|
||
}
|
||
for _, addition := range additions {
|
||
exists, err := r.columnExists(ctx, "lucky_draw_records", addition.name)
|
||
if err != nil {
|
||
return err
|
||
}
|
||
if !exists {
|
||
if _, err := r.db.ExecContext(ctx, addition.sql); err != nil {
|
||
return err
|
||
}
|
||
}
|
||
}
|
||
return nil
|
||
}
|
||
|
||
func (r *Repository) ensureLuckyGiftRuleVersionTables(ctx context.Context) error {
|
||
statements := []string{
|
||
`CREATE TABLE IF NOT EXISTS lucky_gift_rule_versions (
|
||
app_code VARCHAR(32) NOT NULL DEFAULT 'lalu' COMMENT '应用编码',
|
||
pool_id VARCHAR(96) NOT NULL COMMENT '幸运礼物奖池 ID',
|
||
rule_version BIGINT NOT NULL COMMENT '不可变规则版本',
|
||
enabled TINYINT(1) NOT NULL DEFAULT 0 COMMENT '是否启用',
|
||
target_rtp_ppm BIGINT NOT NULL COMMENT '基础返奖目标，ppm',
|
||
pool_rate_ppm BIGINT NOT NULL COMMENT '每笔扣费进入基础奖池比例，ppm',
|
||
settlement_window_wager BIGINT NOT NULL COMMENT 'RTP 结算窗口流水，单位金币',
|
||
control_band_ppm BIGINT NOT NULL COMMENT '正常波动带，ppm',
|
||
gift_price_reference BIGINT NOT NULL COMMENT '配置参考礼物价格',
|
||
novice_max_equivalent_draws BIGINT NOT NULL DEFAULT 2000 COMMENT '新手阶段截止等价抽数，按累计金币/参考价格折算',
|
||
normal_max_equivalent_draws BIGINT NOT NULL DEFAULT 20000 COMMENT '正常阶段截止等价抽数，超过后进入高阶阶段',
|
||
max_single_payout BIGINT NOT NULL DEFAULT 50000 COMMENT '单次基础返奖上限，按参考价格配置',
|
||
user_hourly_payout_cap BIGINT NOT NULL DEFAULT 34200 COMMENT '单用户每小时基础返奖上限，按参考价格配置',
|
||
user_daily_payout_cap BIGINT NOT NULL DEFAULT 615600 COMMENT '单用户每日基础返奖上限，按参考价格配置',
|
||
device_daily_payout_cap BIGINT NOT NULL DEFAULT 1026000 COMMENT '单设备每日基础返奖上限，按参考价格配置',
|
||
room_hourly_payout_cap BIGINT NOT NULL DEFAULT 684000 COMMENT '单房间每小时基础返奖上限，按参考价格配置',
|
||
anchor_daily_payout_cap BIGINT NOT NULL DEFAULT 12312000 COMMENT '单主播每日基础返奖上限，按参考价格配置',
|
||
effective_from_ms BIGINT NOT NULL COMMENT '生效时间，UTC epoch ms',
|
||
created_by_admin_id BIGINT NOT NULL DEFAULT 0 COMMENT '发布人',
|
||
created_at_ms BIGINT NOT NULL COMMENT '创建时间，UTC epoch ms',
|
||
PRIMARY KEY (app_code, pool_id, rule_version),
|
||
KEY idx_lucky_gift_rule_versions_latest (app_code, pool_id, rule_version),
|
||
KEY idx_lucky_gift_rule_versions_enabled (app_code, enabled, created_at_ms)
|
||
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci COMMENT='幸运礼物 v2 规则版本表'`,
|
||
`CREATE TABLE IF NOT EXISTS lucky_gift_stage_tiers (
|
||
app_code VARCHAR(32) NOT NULL DEFAULT 'lalu' COMMENT '应用编码',
|
||
pool_id VARCHAR(96) NOT NULL COMMENT '幸运礼物奖池 ID',
|
||
rule_version BIGINT NOT NULL COMMENT '规则版本',
|
||
stage VARCHAR(32) NOT NULL COMMENT 'novice/normal/advanced',
|
||
tier_id VARCHAR(96) NOT NULL COMMENT '奖档 ID',
|
||
multiplier_ppm BIGINT NOT NULL COMMENT '倍率，1x = 1000000',
|
||
base_weight_ppm BIGINT NOT NULL COMMENT '正常模式基础概率，ppm',
|
||
reward_source VARCHAR(32) NOT NULL COMMENT 'base_rtp/activity_subsidy/presentation_only',
|
||
high_water_only TINYINT(1) NOT NULL DEFAULT 0 COMMENT '是否高水位才开放',
|
||
broadcast_level VARCHAR(32) NOT NULL DEFAULT 'none' COMMENT 'none/room/region/global',
|
||
enabled TINYINT(1) NOT NULL DEFAULT 1 COMMENT '是否启用',
|
||
PRIMARY KEY (app_code, pool_id, rule_version, stage, tier_id),
|
||
KEY idx_lucky_gift_stage_tiers_stage (app_code, pool_id, rule_version, stage)
|
||
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci COMMENT='幸运礼物 v2 阶段奖档表'`,
|
||
}
|
||
for _, statement := range statements {
|
||
if _, err := r.db.ExecContext(ctx, statement); err != nil {
|
||
return err
|
||
}
|
||
}
|
||
if err := r.ensureLuckyGiftRuleVersionStageThresholdColumns(ctx); err != nil {
|
||
return err
|
||
}
|
||
if err := r.ensureLuckyGiftRuleVersionRiskCapColumns(ctx); err != nil {
|
||
return err
|
||
}
|
||
return nil
|
||
}
|
||
|
||
func (r *Repository) ensureLuckyGiftRuleVersionStageThresholdColumns(ctx context.Context) error {
|
||
// 阶段阈值从硬编码抽数升级为“累计金币流水折算等价抽数”；旧开发库重启后必须自动补列，避免线上逻辑读到零阈值。
|
||
additions := []struct {
|
||
name string
|
||
sql string
|
||
}{
|
||
{"novice_max_equivalent_draws", `ALTER TABLE lucky_gift_rule_versions ADD COLUMN novice_max_equivalent_draws BIGINT NOT NULL DEFAULT 2000 COMMENT '新手阶段截止等价抽数，按累计金币/参考价格折算' AFTER gift_price_reference`},
|
||
{"normal_max_equivalent_draws", `ALTER TABLE lucky_gift_rule_versions ADD COLUMN normal_max_equivalent_draws BIGINT NOT NULL DEFAULT 20000 COMMENT '正常阶段截止等价抽数，超过后进入高阶阶段' AFTER novice_max_equivalent_draws`},
|
||
}
|
||
for _, addition := range additions {
|
||
exists, err := r.columnExists(ctx, "lucky_gift_rule_versions", addition.name)
|
||
if err != nil {
|
||
return err
|
||
}
|
||
if !exists {
|
||
if _, err := r.db.ExecContext(ctx, addition.sql); err != nil {
|
||
return err
|
||
}
|
||
}
|
||
}
|
||
return nil
|
||
}
|
||
|
||
func (r *Repository) ensureLuckyGiftRuleVersionRiskCapColumns(ctx context.Context) error {
|
||
// 开发库可能已经存在旧 v2 表；这里做幂等补列，保证服务重启后后台立即可以读写风控上限。
|
||
additions := []struct {
|
||
name string
|
||
sql string
|
||
}{
|
||
{"max_single_payout", `ALTER TABLE lucky_gift_rule_versions ADD COLUMN max_single_payout BIGINT NOT NULL DEFAULT 50000 COMMENT '单次基础返奖上限，按参考价格配置' AFTER gift_price_reference`},
|
||
{"user_hourly_payout_cap", `ALTER TABLE lucky_gift_rule_versions ADD COLUMN user_hourly_payout_cap BIGINT NOT NULL DEFAULT 34200 COMMENT '单用户每小时基础返奖上限，按参考价格配置' AFTER max_single_payout`},
|
||
{"user_daily_payout_cap", `ALTER TABLE lucky_gift_rule_versions ADD COLUMN user_daily_payout_cap BIGINT NOT NULL DEFAULT 615600 COMMENT '单用户每日基础返奖上限，按参考价格配置' AFTER user_hourly_payout_cap`},
|
||
{"device_daily_payout_cap", `ALTER TABLE lucky_gift_rule_versions ADD COLUMN device_daily_payout_cap BIGINT NOT NULL DEFAULT 1026000 COMMENT '单设备每日基础返奖上限，按参考价格配置' AFTER user_daily_payout_cap`},
|
||
{"room_hourly_payout_cap", `ALTER TABLE lucky_gift_rule_versions ADD COLUMN room_hourly_payout_cap BIGINT NOT NULL DEFAULT 684000 COMMENT '单房间每小时基础返奖上限，按参考价格配置' AFTER device_daily_payout_cap`},
|
||
{"anchor_daily_payout_cap", `ALTER TABLE lucky_gift_rule_versions ADD COLUMN anchor_daily_payout_cap BIGINT NOT NULL DEFAULT 12312000 COMMENT '单主播每日基础返奖上限，按参考价格配置' AFTER room_hourly_payout_cap`},
|
||
}
|
||
for _, addition := range additions {
|
||
exists, err := r.columnExists(ctx, "lucky_gift_rule_versions", addition.name)
|
||
if err != nil {
|
||
return err
|
||
}
|
||
if !exists {
|
||
if _, err := r.db.ExecContext(ctx, addition.sql); err != nil {
|
||
return err
|
||
}
|
||
}
|
||
}
|
||
return nil
|
||
}
|
||
|
||
func (r *Repository) ensureDefaultLuckyGiftRules(ctx context.Context) error {
|
||
// 本地和 Docker 开发库通常保留 MySQL volume；这里补齐缺省 lucky/super_lucky 契约，但不覆盖后台已发布规则。
|
||
const seedNowMS int64 = 1779259000000
|
||
const seedTiersJSON = `[
|
||
{"pool":"novice","tier_id":"novice_none","reward_coins":0,"multiplier_ppm":0,"weight":0,"high_water_only":false,"enabled":true},
|
||
{"pool":"novice","tier_id":"novice_1x","reward_coins":100,"multiplier_ppm":1000000,"weight":0,"high_water_only":false,"enabled":true},
|
||
{"pool":"novice","tier_id":"novice_2x","reward_coins":200,"multiplier_ppm":2000000,"weight":0,"high_water_only":false,"enabled":true},
|
||
{"pool":"intermediate","tier_id":"intermediate_none","reward_coins":0,"multiplier_ppm":0,"weight":0,"high_water_only":false,"enabled":true},
|
||
{"pool":"intermediate","tier_id":"intermediate_1x","reward_coins":100,"multiplier_ppm":1000000,"weight":0,"high_water_only":false,"enabled":true},
|
||
{"pool":"intermediate","tier_id":"intermediate_2x","reward_coins":200,"multiplier_ppm":2000000,"weight":0,"high_water_only":false,"enabled":true},
|
||
{"pool":"advanced","tier_id":"advanced_none","reward_coins":0,"multiplier_ppm":0,"weight":0,"high_water_only":false,"enabled":true},
|
||
{"pool":"advanced","tier_id":"advanced_1x","reward_coins":100,"multiplier_ppm":1000000,"weight":0,"high_water_only":false,"enabled":true},
|
||
{"pool":"advanced","tier_id":"advanced_2x","reward_coins":200,"multiplier_ppm":2000000,"weight":0,"high_water_only":false,"enabled":true}
|
||
]`
|
||
_, err := r.db.ExecContext(ctx, `
|
||
INSERT IGNORE INTO lucky_gift_rules (
|
||
app_code, gift_id, enabled, rule_version, gift_price, target_rtp_ppm, pool_rate_ppm,
|
||
global_window_draws, gift_window_draws, novice_draw_limit, intermediate_draw_limit,
|
||
high_multiplier, high_water_pool_multiple, platform_pool_weight_ppm, room_pool_weight_ppm,
|
||
gift_pool_weight_ppm, initial_platform_pool, initial_gift_pool, initial_room_pool,
|
||
platform_reserve, gift_reserve, room_reserve, max_single_payout, user_hourly_payout_cap,
|
||
user_daily_payout_cap, device_daily_payout_cap, room_hourly_payout_cap, anchor_daily_payout_cap,
|
||
room_atmosphere_rate_ppm, room_atmosphere_initial, room_atmosphere_reserve,
|
||
activity_budget, activity_daily_limit, large_tier_enabled, tiers_json,
|
||
updated_by_admin_id, created_at_ms, updated_at_ms
|
||
) VALUES
|
||
('lalu', 'lucky', true, 1, 100, 950000, 950000, 100000, 100000, 2000, 20000, 100, 2, 200000, 300000, 500000, 9500000, 23750000, 5000000, 1000000, 1000000, 100000, 50000, 3420000, 61560000, 102600000, 68400000, 1231200000, 10000, 100000, 10000, 0, 0, true, ?, 0, ?, ?),
|
||
('lalu', 'super_lucky', true, 1, 100, 950000, 950000, 100000, 100000, 2000, 20000, 100, 2, 200000, 300000, 500000, 9500000, 23750000, 5000000, 1000000, 1000000, 100000, 50000, 3420000, 61560000, 102600000, 68400000, 1231200000, 10000, 100000, 10000, 0, 0, true, ?, 0, ?, ?)`,
|
||
seedTiersJSON, seedNowMS, seedNowMS, seedTiersJSON, seedNowMS, seedNowMS,
|
||
)
|
||
return err
|
||
}
|
||
|
||
func (r *Repository) columnExists(ctx context.Context, table string, column string) (bool, error) {
|
||
database, err := r.currentDatabase(ctx)
|
||
if err != nil {
|
||
return false, err
|
||
}
|
||
var count int
|
||
if err := r.db.QueryRowContext(ctx, `
|
||
SELECT COUNT(*)
|
||
FROM information_schema.COLUMNS
|
||
WHERE TABLE_SCHEMA = ? AND TABLE_NAME = ? AND COLUMN_NAME = ?`,
|
||
database, table, column,
|
||
).Scan(&count); err != nil {
|
||
return false, err
|
||
}
|
||
return count > 0, nil
|
||
}
|
||
|
||
func (r *Repository) indexExists(ctx context.Context, table string, index string) (bool, error) {
|
||
database, err := r.currentDatabase(ctx)
|
||
if err != nil {
|
||
return false, err
|
||
}
|
||
var count int
|
||
if err := r.db.QueryRowContext(ctx, `
|
||
SELECT COUNT(*)
|
||
FROM information_schema.STATISTICS
|
||
WHERE TABLE_SCHEMA = ? AND TABLE_NAME = ? AND INDEX_NAME = ?`,
|
||
database, table, index,
|
||
).Scan(&count); err != nil {
|
||
return false, err
|
||
}
|
||
return count > 0, nil
|
||
}
|
||
|
||
func (r *Repository) currentDatabase(ctx context.Context) (string, error) {
|
||
var database sql.NullString
|
||
if err := r.db.QueryRowContext(ctx, `SELECT DATABASE()`).Scan(&database); err != nil {
|
||
return "", err
|
||
}
|
||
if !database.Valid || strings.TrimSpace(database.String) == "" {
|
||
return "", fmt.Errorf("mysql database is not selected")
|
||
}
|
||
return database.String, nil
|
||
}
|
||
|
||
// GetActivity 从 MySQL 活动配置表读取同步查询投影。
|
||
func (r *Repository) GetActivity(ctx context.Context, activityID string) (activitydomain.Activity, error) {
|
||
if r == nil || r.db == nil {
|
||
return activitydomain.Activity{}, xerr.New(xerr.Unavailable, "mysql repository is not configured")
|
||
}
|
||
|
||
row := r.db.QueryRowContext(ctx,
|
||
`SELECT app_code, activity_id, status, updated_at_ms
|
||
FROM activities
|
||
WHERE app_code = ? AND activity_id = ?`,
|
||
appcode.FromContext(ctx),
|
||
activityID,
|
||
)
|
||
|
||
var activity activitydomain.Activity
|
||
var status string
|
||
if err := row.Scan(&activity.AppCode, &activity.ActivityID, &status, &activity.UpdatedAtMs); err != nil {
|
||
if errors.Is(err, sql.ErrNoRows) {
|
||
// 未配置活动时返回 NotFound，上层可决定是否降级为 inactive。
|
||
return activitydomain.Activity{}, xerr.New(xerr.NotFound, "activity not found")
|
||
}
|
||
|
||
return activitydomain.Activity{}, err
|
||
}
|
||
activity.Status = activitydomain.Status(status)
|
||
|
||
return activity, nil
|
||
}
|