1268 lines
53 KiB
Go
1268 lines
53 KiB
Go
package mysql
|
||
|
||
import (
|
||
"context"
|
||
"database/sql"
|
||
"encoding/json"
|
||
"errors"
|
||
"strings"
|
||
"time"
|
||
|
||
mysqlerr "github.com/go-sql-driver/mysql"
|
||
"hyapp/pkg/appcode"
|
||
"hyapp/pkg/xerr"
|
||
)
|
||
|
||
const (
|
||
SourceUser = "user"
|
||
SourceWallet = "wallet"
|
||
SourceRoom = "room"
|
||
SourceGame = "game"
|
||
SourceH5 = "h5"
|
||
)
|
||
|
||
// Repository owns the statistics read model. Online queries must use aggregate
|
||
// tables; the optional activity connection only reads lucky-gift day snapshots,
|
||
// never the high-cardinality draw fact table.
|
||
type Repository struct {
|
||
db *sql.DB
|
||
activityDB *sql.DB
|
||
}
|
||
|
||
func Open(ctx context.Context, dsn string, activityDSNs ...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
|
||
}
|
||
var activityDB *sql.DB
|
||
if len(activityDSNs) > 0 && strings.TrimSpace(activityDSNs[0]) != "" {
|
||
activityDB, err = sql.Open("mysql", strings.TrimSpace(activityDSNs[0]))
|
||
if err != nil {
|
||
_ = db.Close()
|
||
return nil, err
|
||
}
|
||
if err := activityDB.PingContext(ctx); err != nil {
|
||
_ = activityDB.Close()
|
||
_ = db.Close()
|
||
return nil, err
|
||
}
|
||
}
|
||
repo := &Repository{db: db, activityDB: activityDB}
|
||
if err := repo.Migrate(ctx); err != nil {
|
||
if activityDB != nil {
|
||
_ = activityDB.Close()
|
||
}
|
||
_ = db.Close()
|
||
return nil, err
|
||
}
|
||
return repo, nil
|
||
}
|
||
|
||
func (r *Repository) Close() error {
|
||
if r == nil {
|
||
return nil
|
||
}
|
||
var err error
|
||
if r.activityDB != nil {
|
||
err = r.activityDB.Close()
|
||
}
|
||
if r.db != nil {
|
||
if dbErr := r.db.Close(); err == nil {
|
||
err = dbErr
|
||
}
|
||
}
|
||
return err
|
||
}
|
||
|
||
func (r *Repository) Ping(ctx context.Context) error {
|
||
if r == nil || r.db == nil {
|
||
return xerr.New(xerr.Unavailable, "mysql repository is not configured")
|
||
}
|
||
if err := r.db.PingContext(ctx); err != nil {
|
||
return err
|
||
}
|
||
if r.activityDB != nil {
|
||
return r.activityDB.PingContext(ctx)
|
||
}
|
||
return nil
|
||
}
|
||
|
||
func (r *Repository) Migrate(ctx context.Context) error {
|
||
statements := []string{
|
||
`CREATE TABLE IF NOT EXISTS statistics_event_consumption (
|
||
app_code VARCHAR(32) NOT NULL, source VARCHAR(32) NOT NULL, event_id VARCHAR(160) NOT NULL,
|
||
event_type VARCHAR(64) NOT NULL, status VARCHAR(32) NOT NULL, consumed_at_ms BIGINT NOT NULL,
|
||
created_at_ms BIGINT NOT NULL, updated_at_ms BIGINT NOT NULL,
|
||
PRIMARY KEY (app_code, source, event_id),
|
||
KEY idx_statistics_event_type (app_code, source, event_type, consumed_at_ms)
|
||
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci`,
|
||
`CREATE TABLE IF NOT EXISTS stat_app_day_country (
|
||
app_code VARCHAR(32) NOT NULL, stat_day DATE NOT NULL, country_id BIGINT NOT NULL DEFAULT 0,
|
||
region_id BIGINT NOT NULL DEFAULT 0,
|
||
new_users BIGINT NOT NULL DEFAULT 0, active_users BIGINT NOT NULL DEFAULT 0, paid_users BIGINT NOT NULL DEFAULT 0,
|
||
recharge_usd_minor BIGINT NOT NULL DEFAULT 0, new_user_recharge_usd_minor BIGINT NOT NULL DEFAULT 0,
|
||
coin_seller_recharge_usd_minor BIGINT NOT NULL DEFAULT 0, mifapay_recharge_usd_minor BIGINT NOT NULL DEFAULT 0,
|
||
google_recharge_usd_minor BIGINT NOT NULL DEFAULT 0, coin_seller_transfer_coin BIGINT NOT NULL DEFAULT 0,
|
||
gift_coin_spent BIGINT NOT NULL DEFAULT 0,
|
||
lucky_gift_turnover BIGINT NOT NULL DEFAULT 0, lucky_gift_payout BIGINT NOT NULL DEFAULT 0, lucky_gift_payers BIGINT NOT NULL DEFAULT 0,
|
||
game_turnover BIGINT NOT NULL DEFAULT 0, game_players BIGINT NOT NULL DEFAULT 0,
|
||
game_payout BIGINT NOT NULL DEFAULT 0, game_refund BIGINT NOT NULL DEFAULT 0, updated_at_ms BIGINT NOT NULL,
|
||
PRIMARY KEY (app_code, stat_day, country_id, region_id),
|
||
KEY idx_stat_app_day_region (app_code, stat_day, region_id)
|
||
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci`,
|
||
`CREATE TABLE IF NOT EXISTS stat_user_day_activity (
|
||
app_code VARCHAR(32) NOT NULL, stat_day DATE NOT NULL, country_id BIGINT NOT NULL DEFAULT 0,
|
||
region_id BIGINT NOT NULL DEFAULT 0,
|
||
user_id BIGINT NOT NULL, first_active_at_ms BIGINT NOT NULL,
|
||
PRIMARY KEY (app_code, stat_day, user_id),
|
||
KEY idx_stat_user_day_country (app_code, stat_day, country_id),
|
||
KEY idx_stat_user_day_region (app_code, stat_day, region_id)
|
||
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci`,
|
||
`CREATE TABLE IF NOT EXISTS stat_user_registration (
|
||
app_code VARCHAR(32) NOT NULL, user_id BIGINT NOT NULL, registered_day DATE NOT NULL,
|
||
country_id BIGINT NOT NULL DEFAULT 0, region_id BIGINT NOT NULL DEFAULT 0, registered_at_ms BIGINT NOT NULL,
|
||
PRIMARY KEY (app_code, user_id), KEY idx_stat_user_registration_day (app_code, registered_day, country_id),
|
||
KEY idx_stat_user_registration_region (app_code, registered_day, region_id)
|
||
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci`,
|
||
`CREATE TABLE IF NOT EXISTS stat_recharge_day_payers (
|
||
app_code VARCHAR(32) NOT NULL, stat_day DATE NOT NULL, country_id BIGINT NOT NULL DEFAULT 0,
|
||
region_id BIGINT NOT NULL DEFAULT 0,
|
||
user_id BIGINT NOT NULL, first_paid_at_ms BIGINT NOT NULL,
|
||
PRIMARY KEY (app_code, stat_day, user_id), KEY idx_stat_recharge_payer_country (app_code, stat_day, country_id),
|
||
KEY idx_stat_recharge_payer_region (app_code, stat_day, region_id)
|
||
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci`,
|
||
`CREATE TABLE IF NOT EXISTS stat_lucky_gift_day_payers (
|
||
app_code VARCHAR(32) NOT NULL, stat_day DATE NOT NULL, country_id BIGINT NOT NULL DEFAULT 0,
|
||
region_id BIGINT NOT NULL DEFAULT 0,
|
||
user_id BIGINT NOT NULL, first_paid_at_ms BIGINT NOT NULL,
|
||
PRIMARY KEY (app_code, stat_day, user_id),
|
||
KEY idx_stat_lucky_payer_country (app_code, stat_day, country_id),
|
||
KEY idx_stat_lucky_payer_region (app_code, stat_day, region_id)
|
||
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci`,
|
||
`CREATE TABLE IF NOT EXISTS stat_lucky_gift_pool_day_country (
|
||
app_code VARCHAR(32) NOT NULL, stat_day DATE NOT NULL, country_id BIGINT NOT NULL DEFAULT 0,
|
||
region_id BIGINT NOT NULL DEFAULT 0,
|
||
pool_id VARCHAR(96) NOT NULL, turnover_coin BIGINT NOT NULL DEFAULT 0, payout_coin BIGINT NOT NULL DEFAULT 0,
|
||
updated_at_ms BIGINT NOT NULL,
|
||
PRIMARY KEY (app_code, stat_day, country_id, region_id, pool_id),
|
||
KEY idx_stat_lucky_pool_overview (app_code, stat_day, pool_id),
|
||
KEY idx_stat_lucky_pool_region (app_code, stat_day, region_id)
|
||
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci`,
|
||
`CREATE TABLE IF NOT EXISTS stat_real_room_robot_gift_day_rooms (
|
||
app_code VARCHAR(32) NOT NULL, stat_day DATE NOT NULL, country_id BIGINT NOT NULL DEFAULT 0,
|
||
region_id BIGINT NOT NULL DEFAULT 0, room_id VARCHAR(96) NOT NULL,
|
||
normal_gift_coin BIGINT NOT NULL DEFAULT 0,
|
||
lucky_gift_coin BIGINT NOT NULL DEFAULT 0,
|
||
super_lucky_gift_coin BIGINT NOT NULL DEFAULT 0,
|
||
updated_at_ms BIGINT NOT NULL,
|
||
PRIMARY KEY (app_code, stat_day, country_id, region_id, room_id),
|
||
KEY idx_stat_real_room_robot_gift_region (app_code, stat_day, region_id),
|
||
KEY idx_stat_real_room_robot_gift_country (app_code, stat_day, country_id)
|
||
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci`,
|
||
`CREATE TABLE IF NOT EXISTS stat_game_day_country (
|
||
app_code VARCHAR(32) NOT NULL, stat_day DATE NOT NULL, country_id BIGINT NOT NULL DEFAULT 0,
|
||
region_id BIGINT NOT NULL DEFAULT 0,
|
||
platform_code VARCHAR(64) NOT NULL DEFAULT '', game_id VARCHAR(96) NOT NULL,
|
||
turnover_coin BIGINT NOT NULL DEFAULT 0, payout_coin BIGINT NOT NULL DEFAULT 0, refund_coin BIGINT NOT NULL DEFAULT 0,
|
||
player_count BIGINT NOT NULL DEFAULT 0, updated_at_ms BIGINT NOT NULL,
|
||
PRIMARY KEY (app_code, stat_day, country_id, region_id, platform_code, game_id),
|
||
KEY idx_stat_game_day_region (app_code, stat_day, region_id)
|
||
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci`,
|
||
`CREATE TABLE IF NOT EXISTS stat_game_day_players (
|
||
app_code VARCHAR(32) NOT NULL, stat_day DATE NOT NULL, country_id BIGINT NOT NULL DEFAULT 0,
|
||
region_id BIGINT NOT NULL DEFAULT 0,
|
||
platform_code VARCHAR(64) NOT NULL DEFAULT '', game_id VARCHAR(96) NOT NULL,
|
||
user_id BIGINT NOT NULL, first_played_at_ms BIGINT NOT NULL,
|
||
PRIMARY KEY (app_code, stat_day, country_id, platform_code, game_id, user_id),
|
||
KEY idx_stat_game_player_region (app_code, stat_day, region_id)
|
||
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci`,
|
||
`CREATE TABLE IF NOT EXISTS stat_self_game_h5_event_day (
|
||
app_code VARCHAR(32) NOT NULL, stat_day DATE NOT NULL, country_id BIGINT NOT NULL DEFAULT 0,
|
||
region_id BIGINT NOT NULL DEFAULT 0, game_id VARCHAR(96) NOT NULL, event_name VARCHAR(64) NOT NULL,
|
||
language VARCHAR(16) NOT NULL DEFAULT '', client_version VARCHAR(64) NOT NULL DEFAULT '',
|
||
h5_version VARCHAR(64) NOT NULL DEFAULT '', entry_source VARCHAR(64) NOT NULL DEFAULT '',
|
||
event_count BIGINT NOT NULL DEFAULT 0, user_count BIGINT NOT NULL DEFAULT 0,
|
||
success_count BIGINT NOT NULL DEFAULT 0, error_count BIGINT NOT NULL DEFAULT 0,
|
||
duration_ms_sum BIGINT NOT NULL DEFAULT 0, updated_at_ms BIGINT NOT NULL,
|
||
PRIMARY KEY (app_code, stat_day, country_id, region_id, game_id, event_name, language, client_version, h5_version, entry_source),
|
||
KEY idx_stat_self_game_h5_day (app_code, stat_day, game_id, event_name),
|
||
KEY idx_stat_self_game_h5_region (app_code, stat_day, region_id, game_id)
|
||
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci`,
|
||
`CREATE TABLE IF NOT EXISTS stat_self_game_h5_event_users (
|
||
app_code VARCHAR(32) NOT NULL, stat_day DATE NOT NULL, game_id VARCHAR(96) NOT NULL,
|
||
event_name VARCHAR(64) NOT NULL, user_id BIGINT NOT NULL, first_seen_at_ms BIGINT NOT NULL,
|
||
PRIMARY KEY (app_code, stat_day, game_id, event_name, user_id),
|
||
KEY idx_stat_self_game_h5_users_day (app_code, stat_day, game_id, event_name)
|
||
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci`,
|
||
`CREATE TABLE IF NOT EXISTS stat_self_game_match_day (
|
||
app_code VARCHAR(32) NOT NULL, stat_day DATE NOT NULL, country_id BIGINT NOT NULL DEFAULT 0,
|
||
region_id BIGINT NOT NULL DEFAULT 0, game_id VARCHAR(96) NOT NULL, stake_coin BIGINT NOT NULL DEFAULT 0,
|
||
created_matches BIGINT NOT NULL DEFAULT 0, matched_matches BIGINT NOT NULL DEFAULT 0,
|
||
settled_matches BIGINT NOT NULL DEFAULT 0, canceled_matches BIGINT NOT NULL DEFAULT 0,
|
||
failed_matches BIGINT NOT NULL DEFAULT 0, human_matches BIGINT NOT NULL DEFAULT 0,
|
||
robot_matches BIGINT NOT NULL DEFAULT 0, human_participants BIGINT NOT NULL DEFAULT 0,
|
||
robot_participants BIGINT NOT NULL DEFAULT 0, winner_count BIGINT NOT NULL DEFAULT 0,
|
||
loser_count BIGINT NOT NULL DEFAULT 0, draw_count BIGINT NOT NULL DEFAULT 0,
|
||
user_stake_coin BIGINT NOT NULL DEFAULT 0, robot_stake_coin BIGINT NOT NULL DEFAULT 0,
|
||
payout_coin BIGINT NOT NULL DEFAULT 0, refund_coin BIGINT NOT NULL DEFAULT 0,
|
||
platform_profit_coin BIGINT NOT NULL DEFAULT 0, pool_delta_coin BIGINT NOT NULL DEFAULT 0,
|
||
forced_player_lose_matches BIGINT NOT NULL DEFAULT 0, wait_ms_sum BIGINT NOT NULL DEFAULT 0,
|
||
duration_ms_sum BIGINT NOT NULL DEFAULT 0, updated_at_ms BIGINT NOT NULL,
|
||
PRIMARY KEY (app_code, stat_day, country_id, region_id, game_id, stake_coin),
|
||
KEY idx_stat_self_game_match_day (app_code, stat_day, game_id),
|
||
KEY idx_stat_self_game_match_region (app_code, stat_day, region_id, game_id)
|
||
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci`,
|
||
`CREATE TABLE IF NOT EXISTS stat_self_game_pool_day (
|
||
app_code VARCHAR(32) NOT NULL, stat_day DATE NOT NULL, game_id VARCHAR(96) NOT NULL,
|
||
direction VARCHAR(16) NOT NULL DEFAULT '', reason VARCHAR(64) NOT NULL DEFAULT '',
|
||
transaction_count BIGINT NOT NULL DEFAULT 0, in_coin BIGINT NOT NULL DEFAULT 0,
|
||
out_coin BIGINT NOT NULL DEFAULT 0, balance_after_coin BIGINT NOT NULL DEFAULT 0,
|
||
updated_at_ms BIGINT NOT NULL,
|
||
PRIMARY KEY (app_code, stat_day, game_id, direction, reason),
|
||
KEY idx_stat_self_game_pool_day (app_code, stat_day, game_id)
|
||
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci`,
|
||
`CREATE TABLE IF NOT EXISTS stat_self_game_participant_day (
|
||
app_code VARCHAR(32) NOT NULL, stat_day DATE NOT NULL, country_id BIGINT NOT NULL DEFAULT 0,
|
||
region_id BIGINT NOT NULL DEFAULT 0, game_id VARCHAR(96) NOT NULL, stake_coin BIGINT NOT NULL DEFAULT 0,
|
||
participant_type VARCHAR(32) NOT NULL DEFAULT 'user', result VARCHAR(32) NOT NULL DEFAULT '',
|
||
rps_gesture VARCHAR(32) NOT NULL DEFAULT '', dice_point INT NOT NULL DEFAULT 0,
|
||
participant_count BIGINT NOT NULL DEFAULT 0, payout_coin BIGINT NOT NULL DEFAULT 0,
|
||
net_win_coin BIGINT NOT NULL DEFAULT 0, updated_at_ms BIGINT NOT NULL,
|
||
PRIMARY KEY (app_code, stat_day, country_id, region_id, game_id, stake_coin, participant_type, result, rps_gesture, dice_point),
|
||
KEY idx_stat_self_game_participant_day (app_code, stat_day, game_id),
|
||
KEY idx_stat_self_game_participant_region (app_code, stat_day, region_id, game_id)
|
||
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci`,
|
||
`CREATE TABLE IF NOT EXISTS stat_self_game_user_day (
|
||
app_code VARCHAR(32) NOT NULL, stat_day DATE NOT NULL, country_id BIGINT NOT NULL DEFAULT 0,
|
||
region_id BIGINT NOT NULL DEFAULT 0, game_id VARCHAR(96) NOT NULL, user_id BIGINT NOT NULL,
|
||
match_count BIGINT NOT NULL DEFAULT 0, win_count BIGINT NOT NULL DEFAULT 0,
|
||
lose_count BIGINT NOT NULL DEFAULT 0, draw_count BIGINT NOT NULL DEFAULT 0,
|
||
cancel_count BIGINT NOT NULL DEFAULT 0, stake_coin BIGINT NOT NULL DEFAULT 0,
|
||
payout_coin BIGINT NOT NULL DEFAULT 0, net_win_coin BIGINT NOT NULL DEFAULT 0,
|
||
updated_at_ms BIGINT NOT NULL,
|
||
PRIMARY KEY (app_code, stat_day, game_id, user_id),
|
||
KEY idx_stat_self_game_user_win (app_code, stat_day, game_id, net_win_coin),
|
||
KEY idx_stat_self_game_user_region (app_code, stat_day, region_id, game_id)
|
||
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci`,
|
||
}
|
||
for _, statement := range statements {
|
||
if _, err := r.db.ExecContext(ctx, statement); err != nil {
|
||
return err
|
||
}
|
||
}
|
||
alterStatements := []string{
|
||
`ALTER TABLE stat_app_day_country ADD COLUMN paid_users BIGINT NOT NULL DEFAULT 0 AFTER active_users`,
|
||
`ALTER TABLE stat_app_day_country ADD COLUMN game_refund BIGINT NOT NULL DEFAULT 0 AFTER game_payout`,
|
||
`ALTER TABLE stat_app_day_country ADD COLUMN lucky_gift_payout BIGINT NOT NULL DEFAULT 0 AFTER lucky_gift_turnover`,
|
||
`ALTER TABLE stat_app_day_country ADD COLUMN coin_seller_transfer_coin BIGINT NOT NULL DEFAULT 0 AFTER google_recharge_usd_minor`,
|
||
`ALTER TABLE stat_game_day_country ADD COLUMN refund_coin BIGINT NOT NULL DEFAULT 0 AFTER payout_coin`,
|
||
`ALTER TABLE stat_app_day_country ADD COLUMN region_id BIGINT NOT NULL DEFAULT 0 AFTER country_id`,
|
||
`ALTER TABLE stat_user_day_activity ADD COLUMN region_id BIGINT NOT NULL DEFAULT 0 AFTER country_id`,
|
||
`ALTER TABLE stat_user_registration ADD COLUMN region_id BIGINT NOT NULL DEFAULT 0 AFTER country_id`,
|
||
`ALTER TABLE stat_recharge_day_payers ADD COLUMN region_id BIGINT NOT NULL DEFAULT 0 AFTER country_id`,
|
||
`ALTER TABLE stat_lucky_gift_day_payers ADD COLUMN region_id BIGINT NOT NULL DEFAULT 0 AFTER country_id`,
|
||
`ALTER TABLE stat_lucky_gift_pool_day_country ADD COLUMN region_id BIGINT NOT NULL DEFAULT 0 AFTER country_id`,
|
||
`ALTER TABLE stat_game_day_country ADD COLUMN region_id BIGINT NOT NULL DEFAULT 0 AFTER country_id`,
|
||
`ALTER TABLE stat_game_day_players ADD COLUMN region_id BIGINT NOT NULL DEFAULT 0 AFTER country_id`,
|
||
`ALTER TABLE stat_app_day_country ADD KEY idx_stat_app_day_region (app_code, stat_day, region_id)`,
|
||
`ALTER TABLE stat_user_day_activity ADD KEY idx_stat_user_day_region (app_code, stat_day, region_id)`,
|
||
`ALTER TABLE stat_user_registration ADD KEY idx_stat_user_registration_region (app_code, registered_day, region_id)`,
|
||
`ALTER TABLE stat_recharge_day_payers ADD KEY idx_stat_recharge_payer_region (app_code, stat_day, region_id)`,
|
||
`ALTER TABLE stat_lucky_gift_day_payers ADD KEY idx_stat_lucky_payer_region (app_code, stat_day, region_id)`,
|
||
`ALTER TABLE stat_lucky_gift_pool_day_country ADD KEY idx_stat_lucky_pool_region (app_code, stat_day, region_id)`,
|
||
`ALTER TABLE stat_game_day_country ADD KEY idx_stat_game_day_region (app_code, stat_day, region_id)`,
|
||
`ALTER TABLE stat_game_day_players ADD KEY idx_stat_game_player_region (app_code, stat_day, region_id)`,
|
||
}
|
||
for _, statement := range alterStatements {
|
||
if _, err := r.db.ExecContext(ctx, statement); err != nil && !isDuplicateSchemaError(err) {
|
||
return err
|
||
}
|
||
}
|
||
for table, columns := range map[string][]string{
|
||
"stat_app_day_country": {"app_code", "stat_day", "country_id", "region_id"},
|
||
"stat_lucky_gift_pool_day_country": {"app_code", "stat_day", "country_id", "region_id", "pool_id"},
|
||
"stat_game_day_country": {"app_code", "stat_day", "country_id", "region_id", "platform_code", "game_id"},
|
||
} {
|
||
if err := r.ensurePrimaryKeyColumns(ctx, table, columns); err != nil {
|
||
return err
|
||
}
|
||
}
|
||
return nil
|
||
}
|
||
|
||
func (r *Repository) ensurePrimaryKeyColumns(ctx context.Context, table string, expected []string) error {
|
||
rows, err := r.db.QueryContext(ctx, `
|
||
SELECT COLUMN_NAME
|
||
FROM information_schema.KEY_COLUMN_USAGE
|
||
WHERE TABLE_SCHEMA = DATABASE() AND TABLE_NAME = ? AND CONSTRAINT_NAME = 'PRIMARY'
|
||
ORDER BY ORDINAL_POSITION`, table)
|
||
if err != nil {
|
||
return err
|
||
}
|
||
defer rows.Close()
|
||
|
||
actual := make([]string, 0, len(expected))
|
||
for rows.Next() {
|
||
var column string
|
||
if err := rows.Scan(&column); err != nil {
|
||
return err
|
||
}
|
||
actual = append(actual, column)
|
||
}
|
||
if err := rows.Err(); err != nil {
|
||
return err
|
||
}
|
||
if sameStringSlice(actual, expected) {
|
||
return nil
|
||
}
|
||
// 统计聚合表必须同时以国家和区域做主键；否则区域筛选和国家筛选会写入同一行，导致阿富汗这类历史错位国家继续污染榜单。
|
||
_, err = r.db.ExecContext(ctx, "ALTER TABLE "+table+" DROP PRIMARY KEY, ADD PRIMARY KEY ("+strings.Join(expected, ", ")+")")
|
||
return err
|
||
}
|
||
|
||
func sameStringSlice(left, right []string) bool {
|
||
if len(left) != len(right) {
|
||
return false
|
||
}
|
||
for index := range left {
|
||
if left[index] != right[index] {
|
||
return false
|
||
}
|
||
}
|
||
return true
|
||
}
|
||
|
||
type UserRegisteredEvent struct {
|
||
AppCode string
|
||
EventID string
|
||
UserID int64
|
||
CountryID int64
|
||
RegionID int64
|
||
OccurredAtMS int64
|
||
}
|
||
|
||
type RechargeEvent struct {
|
||
AppCode string
|
||
EventID string
|
||
EventType string
|
||
UserID int64
|
||
USDMinor int64
|
||
CoinAmount int64
|
||
RechargeSequence int64
|
||
TargetCountryID int64
|
||
TargetRegionID int64
|
||
RechargeType string
|
||
OccurredAtMS int64
|
||
}
|
||
|
||
type CoinSellerStockEvent struct {
|
||
AppCode string
|
||
EventID string
|
||
SellerUserID int64
|
||
USDMinor int64
|
||
CoinAmount int64
|
||
CountryID int64
|
||
RegionID int64
|
||
OccurredAtMS int64
|
||
}
|
||
|
||
type RoomGiftEvent struct {
|
||
AppCode string
|
||
EventID string
|
||
RoomID string
|
||
SenderUserID int64
|
||
CountryID int64
|
||
VisibleRegionID int64
|
||
GiftValue int64
|
||
CoinSpent int64
|
||
PoolID string
|
||
GiftTypeCode string
|
||
RealRoomRobot bool
|
||
OccurredAtMS int64
|
||
}
|
||
|
||
type LuckyGiftRewardEvent struct {
|
||
AppCode string
|
||
EventID string
|
||
EventType string
|
||
UserID int64
|
||
CountryID int64
|
||
RegionID int64
|
||
PoolID string
|
||
Amount int64
|
||
OccurredAtMS int64
|
||
}
|
||
|
||
type UserActiveEvent struct {
|
||
AppCode string
|
||
EventID string
|
||
EventType string
|
||
UserID int64
|
||
CountryID int64
|
||
RegionID int64
|
||
OccurredAtMS int64
|
||
}
|
||
|
||
type GameOrderEvent struct {
|
||
AppCode string
|
||
EventID string
|
||
UserID int64
|
||
CountryID int64
|
||
RegionID int64
|
||
PlatformCode string
|
||
GameID string
|
||
OpType string
|
||
CoinAmount int64
|
||
OccurredAtMS int64
|
||
}
|
||
|
||
type SelfGameH5Event struct {
|
||
AppCode string
|
||
EventID string
|
||
EventName string
|
||
GameID string
|
||
Page string
|
||
SessionID string
|
||
MatchID string
|
||
UserID int64
|
||
CountryID int64
|
||
RegionID int64
|
||
Language string
|
||
ClientVersion string
|
||
H5Version string
|
||
EntrySource string
|
||
DurationMS int64
|
||
Success bool
|
||
ErrorCode string
|
||
OccurredAtMS int64
|
||
}
|
||
|
||
type SelfGameMatchParticipantEvent struct {
|
||
UserID int64
|
||
ParticipantType string
|
||
StakeCoin int64
|
||
DicePoints []int64
|
||
RPSGesture string
|
||
Result string
|
||
PayoutCoin int64
|
||
NetWinCoin int64
|
||
}
|
||
|
||
type SelfGameMatchEvent struct {
|
||
AppCode string
|
||
EventID string
|
||
EventType string
|
||
MatchID string
|
||
GameID string
|
||
PlatformCode string
|
||
RoomID string
|
||
CountryID int64
|
||
RegionID int64
|
||
StakeCoin int64
|
||
Status string
|
||
Result string
|
||
MatchMode string
|
||
ForcedResult string
|
||
UserParticipants int64
|
||
RobotParticipants int64
|
||
WinnerCount int64
|
||
LoserCount int64
|
||
DrawCount int64
|
||
UserStakeCoin int64
|
||
RobotStakeCoin int64
|
||
PayoutCoin int64
|
||
RefundCoin int64
|
||
PlatformProfitCoin int64
|
||
PoolDeltaCoin int64
|
||
WaitMS int64
|
||
DurationMS int64
|
||
Participants []SelfGameMatchParticipantEvent
|
||
OccurredAtMS int64
|
||
}
|
||
|
||
type SelfGamePoolAdjustmentEvent struct {
|
||
AppCode string
|
||
EventID string
|
||
GameID string
|
||
Direction string
|
||
Reason string
|
||
AmountCoin int64
|
||
BalanceAfter int64
|
||
OccurredAtMS int64
|
||
}
|
||
|
||
func (r *Repository) ConsumeUserRegistered(ctx context.Context, event UserRegisteredEvent) error {
|
||
return r.withEvent(ctx, SourceUser, event.EventID, "UserRegistered", func(tx *sql.Tx, nowMS int64) error {
|
||
day := statDay(event.OccurredAtMS)
|
||
countryID, regionID := normalizeDimension(event.CountryID, event.RegionID)
|
||
if err := ensureAppDay(ctx, tx, event.AppCode, day, countryID, regionID, nowMS); err != nil {
|
||
return err
|
||
}
|
||
inserted, err := insertUnique(ctx, tx, `
|
||
INSERT IGNORE INTO stat_user_registration (app_code, user_id, registered_day, country_id, region_id, registered_at_ms)
|
||
VALUES (?, ?, ?, ?, ?, ?)
|
||
`, appcode.Normalize(event.AppCode), event.UserID, day, countryID, regionID, event.OccurredAtMS)
|
||
if err != nil {
|
||
return err
|
||
}
|
||
// MQ 重投和补偿任务可能带来新的 event_id，但同一用户注册 cohort 只能入库一次；
|
||
// new_users 必须按 INSERT IGNORE 的真实插入行数累加，避免回放把国家新增人数重复放大。
|
||
_, err = tx.ExecContext(ctx, `
|
||
UPDATE stat_app_day_country
|
||
SET new_users = new_users + ?, updated_at_ms = ?
|
||
WHERE app_code = ? AND stat_day = ? AND country_id = ? AND region_id = ?
|
||
`, inserted, nowMS, appcode.Normalize(event.AppCode), day, countryID, regionID)
|
||
return err
|
||
})
|
||
}
|
||
|
||
func (r *Repository) ConsumeRecharge(ctx context.Context, event RechargeEvent) error {
|
||
return r.withEvent(ctx, SourceWallet, event.EventID, event.EventType, func(tx *sql.Tx, nowMS int64) error {
|
||
day := statDay(event.OccurredAtMS)
|
||
countryID, regionID := normalizeDimension(event.TargetCountryID, event.TargetRegionID)
|
||
if err := ensureAppDay(ctx, tx, event.AppCode, day, countryID, regionID, nowMS); err != nil {
|
||
return err
|
||
}
|
||
paidUsers, err := insertUnique(ctx, tx, `
|
||
INSERT IGNORE INTO stat_recharge_day_payers (app_code, stat_day, country_id, region_id, user_id, first_paid_at_ms)
|
||
VALUES (?, ?, ?, ?, ?, ?)
|
||
`, appcode.Normalize(event.AppCode), day, countryID, regionID, event.UserID, event.OccurredAtMS)
|
||
if err != nil {
|
||
return err
|
||
}
|
||
newUserUSD := int64(0)
|
||
if event.RechargeSequence == 1 {
|
||
newUserUSD = event.USDMinor
|
||
}
|
||
coinSellerUSD, mifapayUSD, googleUSD, coinSellerTransferCoin := int64(0), int64(0), int64(0), int64(0)
|
||
switch strings.ToLower(strings.TrimSpace(event.RechargeType)) {
|
||
case "google", "google_play":
|
||
googleUSD = event.USDMinor
|
||
case "mifapay":
|
||
mifapayUSD = event.USDMinor
|
||
case "usdt_trc20":
|
||
// USDT-TRC20 是站外 H5 充值渠道；当前大盘只单列 MiFaPay/Google/币商转账，
|
||
// 因此只进入总充值和付费用户，不误归到币商美元充值桶。
|
||
case "coin_seller_transfer", "coin_seller":
|
||
coinSellerTransferCoin = event.CoinAmount
|
||
default:
|
||
coinSellerUSD = event.USDMinor
|
||
}
|
||
_, err = tx.ExecContext(ctx, `
|
||
UPDATE stat_app_day_country
|
||
SET recharge_usd_minor = recharge_usd_minor + ?,
|
||
new_user_recharge_usd_minor = new_user_recharge_usd_minor + ?,
|
||
coin_seller_recharge_usd_minor = coin_seller_recharge_usd_minor + ?,
|
||
mifapay_recharge_usd_minor = mifapay_recharge_usd_minor + ?,
|
||
google_recharge_usd_minor = google_recharge_usd_minor + ?,
|
||
coin_seller_transfer_coin = coin_seller_transfer_coin + ?,
|
||
paid_users = paid_users + ?,
|
||
updated_at_ms = ?
|
||
WHERE app_code = ? AND stat_day = ? AND country_id = ? AND region_id = ?
|
||
`, event.USDMinor, newUserUSD, coinSellerUSD, mifapayUSD, googleUSD, coinSellerTransferCoin, paidUsers, nowMS, appcode.Normalize(event.AppCode), day, countryID, regionID)
|
||
return err
|
||
})
|
||
}
|
||
|
||
func (r *Repository) ConsumeCoinSellerStock(ctx context.Context, event CoinSellerStockEvent) error {
|
||
return r.withEvent(ctx, SourceWallet, event.EventID, "WalletCoinSellerStockPurchased", func(tx *sql.Tx, nowMS int64) error {
|
||
day := statDay(event.OccurredAtMS)
|
||
countryID, regionID := normalizeDimension(event.CountryID, event.RegionID)
|
||
if err := ensureAppDay(ctx, tx, event.AppCode, day, countryID, regionID, nowMS); err != nil {
|
||
return err
|
||
}
|
||
// 币商进货是后台 USDT 入金事实，只进入总充值和币商充值列；它不是普通用户到账，所以不增加用户 USDT 首充和付费用户。
|
||
_, err := tx.ExecContext(ctx, `
|
||
UPDATE stat_app_day_country
|
||
SET recharge_usd_minor = recharge_usd_minor + ?,
|
||
coin_seller_recharge_usd_minor = coin_seller_recharge_usd_minor + ?,
|
||
updated_at_ms = ?
|
||
WHERE app_code = ? AND stat_day = ? AND country_id = ? AND region_id = ?
|
||
`, event.USDMinor, event.USDMinor, nowMS, appcode.Normalize(event.AppCode), day, countryID, regionID)
|
||
return err
|
||
})
|
||
}
|
||
|
||
func (r *Repository) ConsumeRoomGift(ctx context.Context, event RoomGiftEvent) error {
|
||
return r.withEvent(ctx, SourceRoom, event.EventID, "RoomGiftSent", func(tx *sql.Tx, nowMS int64) error {
|
||
day := statDay(event.OccurredAtMS)
|
||
countryID, regionID := normalizeDimension(event.CountryID, event.VisibleRegionID)
|
||
if event.RealRoomRobot {
|
||
normalCoin, luckyCoin, superLuckyCoin := realRoomRobotGiftCoins(event)
|
||
// 真人房机器人礼物单独进 Databi 投放统计，不混入普通用户付费/幸运礼物流水，避免运营指标被机器行为放大。
|
||
_, err := tx.ExecContext(ctx, `
|
||
INSERT INTO stat_real_room_robot_gift_day_rooms (
|
||
app_code, stat_day, country_id, region_id, room_id,
|
||
normal_gift_coin, lucky_gift_coin, super_lucky_gift_coin, updated_at_ms
|
||
) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)
|
||
ON DUPLICATE KEY UPDATE
|
||
normal_gift_coin = normal_gift_coin + VALUES(normal_gift_coin),
|
||
lucky_gift_coin = lucky_gift_coin + VALUES(lucky_gift_coin),
|
||
super_lucky_gift_coin = super_lucky_gift_coin + VALUES(super_lucky_gift_coin),
|
||
updated_at_ms = VALUES(updated_at_ms)
|
||
`, appcode.Normalize(event.AppCode), day, countryID, regionID, strings.TrimSpace(event.RoomID), normalCoin, luckyCoin, superLuckyCoin, nowMS)
|
||
return err
|
||
}
|
||
if err := ensureAppDay(ctx, tx, event.AppCode, day, countryID, regionID, nowMS); err != nil {
|
||
return err
|
||
}
|
||
luckyTurnover, luckyPayers := int64(0), int64(0)
|
||
poolID := strings.TrimSpace(event.PoolID)
|
||
if poolID != "" {
|
||
luckyTurnover = event.CoinSpent
|
||
affected, err := insertUnique(ctx, tx, `
|
||
INSERT IGNORE INTO stat_lucky_gift_day_payers (app_code, stat_day, country_id, region_id, user_id, first_paid_at_ms)
|
||
VALUES (?, ?, ?, ?, ?, ?)
|
||
`, appcode.Normalize(event.AppCode), day, countryID, regionID, event.SenderUserID, event.OccurredAtMS)
|
||
if err != nil {
|
||
return err
|
||
}
|
||
luckyPayers = affected
|
||
// Pool 明细是 Databi 点击下钻的真实口径：每笔带 pool_id 的幸运礼物送礼先按 UTC 日、国家和奖池累加流水，返奖事件稍后用相同维度补齐 payout。
|
||
if err := ensureLuckyGiftPoolDay(ctx, tx, event.AppCode, day, countryID, regionID, poolID, nowMS); err != nil {
|
||
return err
|
||
}
|
||
if _, err := tx.ExecContext(ctx, `
|
||
UPDATE stat_lucky_gift_pool_day_country
|
||
SET turnover_coin = turnover_coin + ?, updated_at_ms = ?
|
||
WHERE app_code = ? AND stat_day = ? AND country_id = ? AND region_id = ? AND pool_id = ?
|
||
`, luckyTurnover, nowMS, appcode.Normalize(event.AppCode), day, countryID, regionID, poolID); err != nil {
|
||
return err
|
||
}
|
||
}
|
||
_, err := tx.ExecContext(ctx, `
|
||
UPDATE stat_app_day_country
|
||
SET gift_coin_spent = gift_coin_spent + ?, lucky_gift_turnover = lucky_gift_turnover + ?,
|
||
lucky_gift_payers = lucky_gift_payers + ?, updated_at_ms = ?
|
||
WHERE app_code = ? AND stat_day = ? AND country_id = ? AND region_id = ?
|
||
`, event.CoinSpent, luckyTurnover, luckyPayers, nowMS, appcode.Normalize(event.AppCode), day, countryID, regionID)
|
||
return err
|
||
})
|
||
}
|
||
|
||
func realRoomRobotGiftCoins(event RoomGiftEvent) (int64, int64, int64) {
|
||
coin := event.CoinSpent
|
||
if coin <= 0 {
|
||
coin = event.GiftValue
|
||
}
|
||
switch realRoomRobotGiftKind(event) {
|
||
case "super_lucky":
|
||
return 0, 0, coin
|
||
case "lucky":
|
||
return 0, coin, 0
|
||
default:
|
||
return coin, 0, 0
|
||
}
|
||
}
|
||
|
||
func realRoomRobotGiftKind(event RoomGiftEvent) string {
|
||
giftType := strings.ToLower(strings.TrimSpace(event.GiftTypeCode))
|
||
poolID := strings.ToLower(strings.TrimSpace(event.PoolID))
|
||
switch {
|
||
case giftType == "super_lucky" || strings.Contains(poolID, "super_lucky"):
|
||
return "super_lucky"
|
||
case giftType == "lucky" || poolID != "":
|
||
return "lucky"
|
||
default:
|
||
return "normal"
|
||
}
|
||
}
|
||
|
||
func (r *Repository) ConsumeLuckyGiftReward(ctx context.Context, event LuckyGiftRewardEvent) error {
|
||
return r.withEvent(ctx, SourceWallet, event.EventID, "WalletLuckyGiftRewardCredited", func(tx *sql.Tx, nowMS int64) error {
|
||
day := statDay(event.OccurredAtMS)
|
||
countryID, regionID := normalizeDimension(event.CountryID, event.RegionID)
|
||
if err := ensureAppDay(ctx, tx, event.AppCode, day, countryID, regionID, nowMS); err != nil {
|
||
return err
|
||
}
|
||
poolID := strings.TrimSpace(event.PoolID)
|
||
if poolID != "" {
|
||
// 钱包返奖事件带回 pool_id 后，按同一 UTC 日和国家写入奖池返奖；利润不落表，查询时用 turnover-payout 实时计算，避免双写漂移。
|
||
if err := ensureLuckyGiftPoolDay(ctx, tx, event.AppCode, day, countryID, regionID, poolID, nowMS); err != nil {
|
||
return err
|
||
}
|
||
if _, err := tx.ExecContext(ctx, `
|
||
UPDATE stat_lucky_gift_pool_day_country
|
||
SET payout_coin = payout_coin + ?, updated_at_ms = ?
|
||
WHERE app_code = ? AND stat_day = ? AND country_id = ? AND region_id = ? AND pool_id = ?
|
||
`, event.Amount, nowMS, appcode.Normalize(event.AppCode), day, countryID, regionID, poolID); err != nil {
|
||
return err
|
||
}
|
||
}
|
||
_, err := tx.ExecContext(ctx, `
|
||
UPDATE stat_app_day_country
|
||
SET lucky_gift_payout = lucky_gift_payout + ?, updated_at_ms = ?
|
||
WHERE app_code = ? AND stat_day = ? AND country_id = ? AND region_id = ?
|
||
`, event.Amount, nowMS, appcode.Normalize(event.AppCode), day, countryID, regionID)
|
||
return err
|
||
})
|
||
}
|
||
|
||
func (r *Repository) ConsumeUserActive(ctx context.Context, event UserActiveEvent) error {
|
||
return r.withEvent(ctx, SourceRoom, event.EventID, event.EventType, func(tx *sql.Tx, nowMS int64) error {
|
||
countryID, regionID := normalizeDimension(event.CountryID, event.RegionID)
|
||
return applyActive(ctx, tx, event.AppCode, statDay(event.OccurredAtMS), countryID, regionID, event.UserID, event.OccurredAtMS, nowMS)
|
||
})
|
||
}
|
||
|
||
func (r *Repository) ConsumeGameOrder(ctx context.Context, event GameOrderEvent) error {
|
||
return r.withEvent(ctx, SourceGame, event.EventID, "GameOrderSettled", func(tx *sql.Tx, nowMS int64) error {
|
||
day := statDay(event.OccurredAtMS)
|
||
countryID, regionID := normalizeDimension(event.CountryID, event.RegionID)
|
||
if err := applyActive(ctx, tx, event.AppCode, day, countryID, regionID, event.UserID, event.OccurredAtMS, nowMS); err != nil {
|
||
return err
|
||
}
|
||
if err := ensureAppDay(ctx, tx, event.AppCode, day, countryID, regionID, nowMS); err != nil {
|
||
return err
|
||
}
|
||
turnover, payout, refund := int64(0), int64(0), int64(0)
|
||
switch strings.ToLower(strings.TrimSpace(event.OpType)) {
|
||
case "debit", "bet":
|
||
turnover = event.CoinAmount
|
||
case "credit", "payout":
|
||
payout = event.CoinAmount
|
||
case "refund", "reverse":
|
||
refund = event.CoinAmount
|
||
}
|
||
playerDelta := int64(0)
|
||
if turnover > 0 {
|
||
affected, err := insertUnique(ctx, tx, `
|
||
INSERT IGNORE INTO stat_game_day_players (app_code, stat_day, country_id, region_id, platform_code, game_id, user_id, first_played_at_ms)
|
||
VALUES (?, ?, ?, ?, ?, ?, ?, ?)
|
||
`, appcode.Normalize(event.AppCode), day, countryID, regionID, event.PlatformCode, event.GameID, event.UserID, event.OccurredAtMS)
|
||
if err != nil {
|
||
return err
|
||
}
|
||
playerDelta = affected
|
||
}
|
||
if err := ensureGameDay(ctx, tx, event.AppCode, day, countryID, regionID, event.PlatformCode, event.GameID, nowMS); err != nil {
|
||
return err
|
||
}
|
||
if _, err := tx.ExecContext(ctx, `
|
||
UPDATE stat_game_day_country
|
||
SET turnover_coin = turnover_coin + ?, payout_coin = payout_coin + ?, refund_coin = refund_coin + ?,
|
||
player_count = player_count + ?, updated_at_ms = ?
|
||
WHERE app_code = ? AND stat_day = ? AND country_id = ? AND region_id = ? AND platform_code = ? AND game_id = ?
|
||
`, turnover, payout, refund, playerDelta, nowMS, appcode.Normalize(event.AppCode), day, countryID, regionID, event.PlatformCode, event.GameID); err != nil {
|
||
return err
|
||
}
|
||
_, err := tx.ExecContext(ctx, `
|
||
UPDATE stat_app_day_country
|
||
SET game_turnover = game_turnover + ?, game_payout = game_payout + ?, game_refund = game_refund + ?,
|
||
game_players = game_players + ?, updated_at_ms = ?
|
||
WHERE app_code = ? AND stat_day = ? AND country_id = ? AND region_id = ?
|
||
`, turnover, payout, refund, playerDelta, nowMS, appcode.Normalize(event.AppCode), day, countryID, regionID)
|
||
return err
|
||
})
|
||
}
|
||
|
||
func (r *Repository) ConsumeSelfGameH5Event(ctx context.Context, event SelfGameH5Event) error {
|
||
eventName := normalizeShortText(event.EventName)
|
||
if eventName == "" {
|
||
eventName = "unknown"
|
||
}
|
||
return r.withEvent(ctx, SourceH5, event.EventID, eventName, func(tx *sql.Tx, nowMS int64) error {
|
||
day := statDay(event.OccurredAtMS)
|
||
countryID, regionID := normalizeDimension(event.CountryID, event.RegionID)
|
||
gameID := normalizeShortText(event.GameID)
|
||
if gameID == "" {
|
||
gameID = "unknown"
|
||
}
|
||
language := normalizeShortText(event.Language)
|
||
clientVersion := normalizeShortText(event.ClientVersion)
|
||
h5Version := normalizeShortText(event.H5Version)
|
||
entrySource := normalizeShortText(event.EntrySource)
|
||
if err := ensureSelfGameH5EventDay(ctx, tx, event.AppCode, day, countryID, regionID, gameID, eventName, language, clientVersion, h5Version, entrySource, nowMS); err != nil {
|
||
return err
|
||
}
|
||
userDelta := int64(0)
|
||
if event.UserID > 0 {
|
||
affected, err := insertUnique(ctx, tx, `
|
||
INSERT IGNORE INTO stat_self_game_h5_event_users (app_code, stat_day, game_id, event_name, user_id, first_seen_at_ms)
|
||
VALUES (?, ?, ?, ?, ?, ?)
|
||
`, appcode.Normalize(event.AppCode), day, gameID, eventName, event.UserID, event.OccurredAtMS)
|
||
if err != nil {
|
||
return err
|
||
}
|
||
userDelta = affected
|
||
}
|
||
successDelta, errorDelta := int64(0), int64(0)
|
||
if event.Success {
|
||
successDelta = 1
|
||
}
|
||
if normalizeShortText(event.ErrorCode) != "" {
|
||
errorDelta = 1
|
||
}
|
||
// H5 事件按 event_id 做幂等；PV 用事件次数，UV 用同日同玩法同事件的用户去重，接口耗时只累加客户端实际传入的 duration。
|
||
_, err := tx.ExecContext(ctx, `
|
||
UPDATE stat_self_game_h5_event_day
|
||
SET event_count = event_count + 1,
|
||
user_count = user_count + ?,
|
||
success_count = success_count + ?,
|
||
error_count = error_count + ?,
|
||
duration_ms_sum = duration_ms_sum + ?,
|
||
updated_at_ms = ?
|
||
WHERE app_code = ? AND stat_day = ? AND country_id = ? AND region_id = ?
|
||
AND game_id = ? AND event_name = ? AND language = ? AND client_version = ?
|
||
AND h5_version = ? AND entry_source = ?
|
||
`, userDelta, successDelta, errorDelta, normalizeID(event.DurationMS), nowMS, appcode.Normalize(event.AppCode), day, countryID, regionID, gameID, eventName, language, clientVersion, h5Version, entrySource)
|
||
return err
|
||
})
|
||
}
|
||
|
||
func (r *Repository) ConsumeSelfGameMatch(ctx context.Context, event SelfGameMatchEvent) error {
|
||
return r.withEvent(ctx, SourceGame, event.EventID, event.EventType, func(tx *sql.Tx, nowMS int64) error {
|
||
day := statDay(event.OccurredAtMS)
|
||
countryID, regionID := normalizeDimension(event.CountryID, event.RegionID)
|
||
gameID := normalizeShortText(event.GameID)
|
||
if gameID == "" {
|
||
gameID = "unknown"
|
||
}
|
||
stakeCoin := normalizeID(event.StakeCoin)
|
||
if err := ensureSelfGameMatchDay(ctx, tx, event.AppCode, day, countryID, regionID, gameID, stakeCoin, nowMS); err != nil {
|
||
return err
|
||
}
|
||
created, matched, settled, canceled, failed := selfGameMatchCounters(event.EventType)
|
||
humanMatch, robotMatch := int64(0), int64(0)
|
||
humanParticipants, robotParticipants := int64(0), int64(0)
|
||
userStake, robotStake, payout, refund, platformProfit, poolDelta := int64(0), int64(0), int64(0), int64(0), int64(0), int64(0)
|
||
winners, losers, draws := int64(0), int64(0), int64(0)
|
||
forcedLose, waitMS, durationMS := int64(0), int64(0), int64(0)
|
||
if settled > 0 {
|
||
if event.UserParticipants > 0 {
|
||
humanMatch = 1
|
||
}
|
||
if event.RobotParticipants > 0 {
|
||
robotMatch = 1
|
||
}
|
||
humanParticipants = normalizeID(event.UserParticipants)
|
||
robotParticipants = normalizeID(event.RobotParticipants)
|
||
winners = normalizeID(event.WinnerCount)
|
||
losers = normalizeID(event.LoserCount)
|
||
draws = normalizeID(event.DrawCount)
|
||
userStake = normalizeID(event.UserStakeCoin)
|
||
robotStake = normalizeID(event.RobotStakeCoin)
|
||
payout = normalizeID(event.PayoutCoin)
|
||
refund = normalizeID(event.RefundCoin)
|
||
platformProfit = event.PlatformProfitCoin
|
||
poolDelta = event.PoolDeltaCoin
|
||
waitMS = normalizeID(event.WaitMS)
|
||
durationMS = normalizeID(event.DurationMS)
|
||
if strings.EqualFold(strings.TrimSpace(event.ForcedResult), "player_lose") {
|
||
forcedLose = 1
|
||
}
|
||
}
|
||
if canceled > 0 || failed > 0 || matched > 0 {
|
||
waitMS = normalizeID(event.WaitMS)
|
||
durationMS = normalizeID(event.DurationMS)
|
||
}
|
||
// 对局状态事件只按 match 级事实累加一次；明细用户维度只在结算/取消时落表，避免创建和 ready 阶段重复计算下注与输赢。
|
||
if _, err := tx.ExecContext(ctx, `
|
||
UPDATE stat_self_game_match_day
|
||
SET created_matches = created_matches + ?,
|
||
matched_matches = matched_matches + ?,
|
||
settled_matches = settled_matches + ?,
|
||
canceled_matches = canceled_matches + ?,
|
||
failed_matches = failed_matches + ?,
|
||
human_matches = human_matches + ?,
|
||
robot_matches = robot_matches + ?,
|
||
human_participants = human_participants + ?,
|
||
robot_participants = robot_participants + ?,
|
||
winner_count = winner_count + ?,
|
||
loser_count = loser_count + ?,
|
||
draw_count = draw_count + ?,
|
||
user_stake_coin = user_stake_coin + ?,
|
||
robot_stake_coin = robot_stake_coin + ?,
|
||
payout_coin = payout_coin + ?,
|
||
refund_coin = refund_coin + ?,
|
||
platform_profit_coin = platform_profit_coin + ?,
|
||
pool_delta_coin = pool_delta_coin + ?,
|
||
forced_player_lose_matches = forced_player_lose_matches + ?,
|
||
wait_ms_sum = wait_ms_sum + ?,
|
||
duration_ms_sum = duration_ms_sum + ?,
|
||
updated_at_ms = ?
|
||
WHERE app_code = ? AND stat_day = ? AND country_id = ? AND region_id = ?
|
||
AND game_id = ? AND stake_coin = ?
|
||
`, created, matched, settled, canceled, failed, humanMatch, robotMatch, humanParticipants, robotParticipants, winners, losers, draws, userStake, robotStake, payout, refund, platformProfit, poolDelta, forcedLose, waitMS, durationMS, nowMS, appcode.Normalize(event.AppCode), day, countryID, regionID, gameID, stakeCoin); err != nil {
|
||
return err
|
||
}
|
||
if settled > 0 {
|
||
for _, participant := range event.Participants {
|
||
if err := consumeSelfGameSettledParticipant(ctx, tx, event, participant, day, countryID, regionID, gameID, stakeCoin, nowMS); err != nil {
|
||
return err
|
||
}
|
||
}
|
||
}
|
||
if canceled > 0 {
|
||
for _, participant := range event.Participants {
|
||
if participant.UserID <= 0 || !strings.EqualFold(strings.TrimSpace(participant.ParticipantType), "user") {
|
||
continue
|
||
}
|
||
if err := ensureSelfGameUserDay(ctx, tx, event.AppCode, day, countryID, regionID, gameID, participant.UserID, nowMS); err != nil {
|
||
return err
|
||
}
|
||
if _, err := tx.ExecContext(ctx, `
|
||
UPDATE stat_self_game_user_day
|
||
SET cancel_count = cancel_count + 1, updated_at_ms = ?
|
||
WHERE app_code = ? AND stat_day = ? AND game_id = ? AND user_id = ?
|
||
`, nowMS, appcode.Normalize(event.AppCode), day, gameID, participant.UserID); err != nil {
|
||
return err
|
||
}
|
||
}
|
||
}
|
||
return nil
|
||
})
|
||
}
|
||
|
||
func (r *Repository) ConsumeSelfGamePoolAdjustment(ctx context.Context, event SelfGamePoolAdjustmentEvent) error {
|
||
return r.withEvent(ctx, SourceGame, event.EventID, "SelfGamePoolAdjusted", func(tx *sql.Tx, nowMS int64) error {
|
||
day := statDay(event.OccurredAtMS)
|
||
gameID := normalizeShortText(event.GameID)
|
||
if gameID == "" {
|
||
gameID = "unknown"
|
||
}
|
||
direction := normalizeShortText(event.Direction)
|
||
reason := normalizeShortText(event.Reason)
|
||
if err := ensureSelfGamePoolDay(ctx, tx, event.AppCode, day, gameID, direction, reason, nowMS); err != nil {
|
||
return err
|
||
}
|
||
inCoin, outCoin := int64(0), int64(0)
|
||
switch direction {
|
||
case "in":
|
||
inCoin = normalizeID(event.AmountCoin)
|
||
case "out":
|
||
outCoin = normalizeID(event.AmountCoin)
|
||
}
|
||
_, err := tx.ExecContext(ctx, `
|
||
UPDATE stat_self_game_pool_day
|
||
SET transaction_count = transaction_count + 1,
|
||
in_coin = in_coin + ?,
|
||
out_coin = out_coin + ?,
|
||
balance_after_coin = ?,
|
||
updated_at_ms = ?
|
||
WHERE app_code = ? AND stat_day = ? AND game_id = ? AND direction = ? AND reason = ?
|
||
`, inCoin, outCoin, event.BalanceAfter, nowMS, appcode.Normalize(event.AppCode), day, gameID, direction, reason)
|
||
return err
|
||
})
|
||
}
|
||
|
||
func consumeSelfGameSettledParticipant(ctx context.Context, tx *sql.Tx, event SelfGameMatchEvent, participant SelfGameMatchParticipantEvent, day string, countryID int64, regionID int64, gameID string, stakeCoin int64, nowMS int64) error {
|
||
participantType := normalizeShortText(participant.ParticipantType)
|
||
if participantType == "" {
|
||
participantType = "user"
|
||
}
|
||
result := normalizeShortText(participant.Result)
|
||
gesture := normalizeShortText(participant.RPSGesture)
|
||
points := participant.DicePoints
|
||
if len(points) == 0 {
|
||
points = []int64{0}
|
||
}
|
||
for _, point := range points {
|
||
dicePoint := int64(0)
|
||
if point > 0 {
|
||
dicePoint = point
|
||
}
|
||
if err := ensureSelfGameParticipantDay(ctx, tx, event.AppCode, day, countryID, regionID, gameID, stakeCoin, participantType, result, gesture, dicePoint, nowMS); err != nil {
|
||
return err
|
||
}
|
||
if _, err := tx.ExecContext(ctx, `
|
||
UPDATE stat_self_game_participant_day
|
||
SET participant_count = participant_count + 1,
|
||
payout_coin = payout_coin + ?,
|
||
net_win_coin = net_win_coin + ?,
|
||
updated_at_ms = ?
|
||
WHERE app_code = ? AND stat_day = ? AND country_id = ? AND region_id = ?
|
||
AND game_id = ? AND stake_coin = ? AND participant_type = ? AND result = ?
|
||
AND rps_gesture = ? AND dice_point = ?
|
||
`, normalizeID(participant.PayoutCoin), participant.NetWinCoin, nowMS, appcode.Normalize(event.AppCode), day, countryID, regionID, gameID, stakeCoin, participantType, result, gesture, dicePoint); err != nil {
|
||
return err
|
||
}
|
||
}
|
||
if participant.UserID <= 0 || participantType != "user" {
|
||
return nil
|
||
}
|
||
if err := ensureSelfGameUserDay(ctx, tx, event.AppCode, day, countryID, regionID, gameID, participant.UserID, nowMS); err != nil {
|
||
return err
|
||
}
|
||
win, lose, draw := int64(0), int64(0), int64(0)
|
||
switch result {
|
||
case "win":
|
||
win = 1
|
||
case "lose":
|
||
lose = 1
|
||
case "draw":
|
||
draw = 1
|
||
}
|
||
_, err := tx.ExecContext(ctx, `
|
||
UPDATE stat_self_game_user_day
|
||
SET match_count = match_count + 1,
|
||
win_count = win_count + ?,
|
||
lose_count = lose_count + ?,
|
||
draw_count = draw_count + ?,
|
||
stake_coin = stake_coin + ?,
|
||
payout_coin = payout_coin + ?,
|
||
net_win_coin = net_win_coin + ?,
|
||
updated_at_ms = ?
|
||
WHERE app_code = ? AND stat_day = ? AND game_id = ? AND user_id = ?
|
||
`, win, lose, draw, normalizeID(participant.StakeCoin), normalizeID(participant.PayoutCoin), participant.NetWinCoin, nowMS, appcode.Normalize(event.AppCode), day, gameID, participant.UserID)
|
||
return err
|
||
}
|
||
|
||
func selfGameMatchCounters(eventType string) (created, matched, settled, canceled, failed int64) {
|
||
switch strings.TrimSpace(eventType) {
|
||
case "SelfGameMatchCreated":
|
||
created = 1
|
||
case "SelfGameMatchReady":
|
||
matched = 1
|
||
case "SelfGameMatchSettled":
|
||
settled = 1
|
||
case "SelfGameMatchCanceled":
|
||
canceled = 1
|
||
case "SelfGameMatchFailed":
|
||
failed = 1
|
||
}
|
||
return created, matched, settled, canceled, failed
|
||
}
|
||
|
||
func (r *Repository) withEvent(ctx context.Context, source string, eventID string, eventType string, apply func(tx *sql.Tx, nowMS int64) error) error {
|
||
var lastErr error
|
||
for attempt := 0; attempt < 5; attempt++ {
|
||
err := r.withEventOnce(ctx, source, eventID, eventType, apply)
|
||
if err == nil {
|
||
return nil
|
||
}
|
||
if !isRetryableMySQLError(err) {
|
||
return err
|
||
}
|
||
lastErr = err
|
||
timer := time.NewTimer(time.Duration(attempt+1) * 50 * time.Millisecond)
|
||
select {
|
||
case <-ctx.Done():
|
||
timer.Stop()
|
||
return ctx.Err()
|
||
case <-timer.C:
|
||
}
|
||
}
|
||
return lastErr
|
||
}
|
||
|
||
func (r *Repository) withEventOnce(ctx context.Context, source string, eventID string, eventType string, apply func(tx *sql.Tx, nowMS int64) error) error {
|
||
if r == nil || r.db == nil {
|
||
return xerr.New(xerr.Unavailable, "mysql repository is not configured")
|
||
}
|
||
app := appcode.FromContext(ctx)
|
||
nowMS := time.Now().UTC().UnixMilli()
|
||
tx, err := r.db.BeginTx(ctx, nil)
|
||
if err != nil {
|
||
return err
|
||
}
|
||
defer func() { _ = tx.Rollback() }()
|
||
result, err := tx.ExecContext(ctx, `
|
||
INSERT IGNORE INTO statistics_event_consumption (app_code, source, event_id, event_type, status, consumed_at_ms, created_at_ms, updated_at_ms)
|
||
VALUES (?, ?, ?, ?, 'processing', ?, ?, ?)
|
||
`, app, source, eventID, eventType, nowMS, nowMS, nowMS)
|
||
if err != nil {
|
||
return err
|
||
}
|
||
affected, _ := result.RowsAffected()
|
||
if affected == 0 {
|
||
return tx.Commit()
|
||
}
|
||
if err := apply(tx, nowMS); err != nil {
|
||
return err
|
||
}
|
||
if _, err := tx.ExecContext(ctx, `
|
||
UPDATE statistics_event_consumption
|
||
SET status = 'done', consumed_at_ms = ?, updated_at_ms = ?
|
||
WHERE app_code = ? AND source = ? AND event_id = ?
|
||
`, nowMS, nowMS, app, source, eventID); err != nil {
|
||
return err
|
||
}
|
||
return tx.Commit()
|
||
}
|
||
|
||
func isRetryableMySQLError(err error) bool {
|
||
var mysqlErr *mysqlerr.MySQLError
|
||
if !errors.As(err, &mysqlErr) {
|
||
return false
|
||
}
|
||
return mysqlErr.Number == 1213 || mysqlErr.Number == 1205
|
||
}
|
||
|
||
func isDuplicateSchemaError(err error) bool {
|
||
var mysqlErr *mysqlerr.MySQLError
|
||
return errors.As(err, &mysqlErr) && (mysqlErr.Number == 1060 || mysqlErr.Number == 1061)
|
||
}
|
||
|
||
func isMissingTableError(err error) bool {
|
||
var mysqlErr *mysqlerr.MySQLError
|
||
return errors.As(err, &mysqlErr) && mysqlErr.Number == 1146
|
||
}
|
||
|
||
func applyActive(ctx context.Context, tx *sql.Tx, app string, day string, countryID int64, regionID int64, userID int64, occurredAtMS int64, nowMS int64) error {
|
||
if err := ensureAppDay(ctx, tx, app, day, countryID, regionID, nowMS); err != nil {
|
||
return err
|
||
}
|
||
affected, err := insertUnique(ctx, tx, `
|
||
INSERT IGNORE INTO stat_user_day_activity (app_code, stat_day, country_id, region_id, user_id, first_active_at_ms)
|
||
VALUES (?, ?, ?, ?, ?, ?)
|
||
`, appcode.Normalize(app), day, countryID, regionID, userID, occurredAtMS)
|
||
if err != nil {
|
||
return err
|
||
}
|
||
if affected == 0 {
|
||
return nil
|
||
}
|
||
_, err = tx.ExecContext(ctx, `
|
||
UPDATE stat_app_day_country
|
||
SET active_users = active_users + 1, updated_at_ms = ?
|
||
WHERE app_code = ? AND stat_day = ? AND country_id = ? AND region_id = ?
|
||
`, nowMS, appcode.Normalize(app), day, countryID, regionID)
|
||
return err
|
||
}
|
||
|
||
func ensureAppDay(ctx context.Context, tx *sql.Tx, app string, day string, countryID int64, regionID int64, nowMS int64) error {
|
||
_, err := tx.ExecContext(ctx, `
|
||
INSERT INTO stat_app_day_country (app_code, stat_day, country_id, region_id, updated_at_ms)
|
||
VALUES (?, ?, ?, ?, ?)
|
||
ON DUPLICATE KEY UPDATE
|
||
updated_at_ms = VALUES(updated_at_ms)
|
||
`, appcode.Normalize(app), day, countryID, regionID, nowMS)
|
||
return err
|
||
}
|
||
|
||
func ensureGameDay(ctx context.Context, tx *sql.Tx, app string, day string, countryID int64, regionID int64, platformCode string, gameID string, nowMS int64) error {
|
||
_, err := tx.ExecContext(ctx, `
|
||
INSERT INTO stat_game_day_country (app_code, stat_day, country_id, region_id, platform_code, game_id, updated_at_ms)
|
||
VALUES (?, ?, ?, ?, ?, ?, ?)
|
||
ON DUPLICATE KEY UPDATE
|
||
updated_at_ms = VALUES(updated_at_ms)
|
||
`, appcode.Normalize(app), day, countryID, regionID, platformCode, gameID, nowMS)
|
||
return err
|
||
}
|
||
|
||
func ensureLuckyGiftPoolDay(ctx context.Context, tx *sql.Tx, app string, day string, countryID int64, regionID int64, poolID string, nowMS int64) error {
|
||
_, err := tx.ExecContext(ctx, `
|
||
INSERT INTO stat_lucky_gift_pool_day_country (app_code, stat_day, country_id, region_id, pool_id, updated_at_ms)
|
||
VALUES (?, ?, ?, ?, ?, ?)
|
||
ON DUPLICATE KEY UPDATE
|
||
updated_at_ms = VALUES(updated_at_ms)
|
||
`, appcode.Normalize(app), day, countryID, regionID, strings.TrimSpace(poolID), nowMS)
|
||
return err
|
||
}
|
||
|
||
func ensureSelfGameH5EventDay(ctx context.Context, tx *sql.Tx, app string, day string, countryID int64, regionID int64, gameID string, eventName string, language string, clientVersion string, h5Version string, entrySource string, nowMS int64) error {
|
||
_, err := tx.ExecContext(ctx, `
|
||
INSERT INTO stat_self_game_h5_event_day (app_code, stat_day, country_id, region_id, game_id, event_name, language, client_version, h5_version, entry_source, updated_at_ms)
|
||
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
|
||
ON DUPLICATE KEY UPDATE
|
||
updated_at_ms = VALUES(updated_at_ms)
|
||
`, appcode.Normalize(app), day, countryID, regionID, gameID, eventName, language, clientVersion, h5Version, entrySource, nowMS)
|
||
return err
|
||
}
|
||
|
||
func ensureSelfGameMatchDay(ctx context.Context, tx *sql.Tx, app string, day string, countryID int64, regionID int64, gameID string, stakeCoin int64, nowMS int64) error {
|
||
_, err := tx.ExecContext(ctx, `
|
||
INSERT INTO stat_self_game_match_day (app_code, stat_day, country_id, region_id, game_id, stake_coin, updated_at_ms)
|
||
VALUES (?, ?, ?, ?, ?, ?, ?)
|
||
ON DUPLICATE KEY UPDATE
|
||
updated_at_ms = VALUES(updated_at_ms)
|
||
`, appcode.Normalize(app), day, countryID, regionID, gameID, stakeCoin, nowMS)
|
||
return err
|
||
}
|
||
|
||
func ensureSelfGamePoolDay(ctx context.Context, tx *sql.Tx, app string, day string, gameID string, direction string, reason string, nowMS int64) error {
|
||
_, err := tx.ExecContext(ctx, `
|
||
INSERT INTO stat_self_game_pool_day (app_code, stat_day, game_id, direction, reason, updated_at_ms)
|
||
VALUES (?, ?, ?, ?, ?, ?)
|
||
ON DUPLICATE KEY UPDATE
|
||
updated_at_ms = VALUES(updated_at_ms)
|
||
`, appcode.Normalize(app), day, gameID, direction, reason, nowMS)
|
||
return err
|
||
}
|
||
|
||
func ensureSelfGameParticipantDay(ctx context.Context, tx *sql.Tx, app string, day string, countryID int64, regionID int64, gameID string, stakeCoin int64, participantType string, result string, gesture string, dicePoint int64, nowMS int64) error {
|
||
_, err := tx.ExecContext(ctx, `
|
||
INSERT INTO stat_self_game_participant_day (app_code, stat_day, country_id, region_id, game_id, stake_coin, participant_type, result, rps_gesture, dice_point, updated_at_ms)
|
||
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
|
||
ON DUPLICATE KEY UPDATE
|
||
updated_at_ms = VALUES(updated_at_ms)
|
||
`, appcode.Normalize(app), day, countryID, regionID, gameID, stakeCoin, participantType, result, gesture, dicePoint, nowMS)
|
||
return err
|
||
}
|
||
|
||
func ensureSelfGameUserDay(ctx context.Context, tx *sql.Tx, app string, day string, countryID int64, regionID int64, gameID string, userID int64, nowMS int64) error {
|
||
_, err := tx.ExecContext(ctx, `
|
||
INSERT INTO stat_self_game_user_day (app_code, stat_day, country_id, region_id, game_id, user_id, updated_at_ms)
|
||
VALUES (?, ?, ?, ?, ?, ?, ?)
|
||
ON DUPLICATE KEY UPDATE
|
||
updated_at_ms = VALUES(updated_at_ms)
|
||
`, appcode.Normalize(app), day, countryID, regionID, gameID, userID, nowMS)
|
||
return err
|
||
}
|
||
|
||
func insertUnique(ctx context.Context, tx *sql.Tx, query string, args ...any) (int64, error) {
|
||
result, err := tx.ExecContext(ctx, query, args...)
|
||
if err != nil {
|
||
return 0, err
|
||
}
|
||
affected, err := result.RowsAffected()
|
||
if err != nil {
|
||
return 0, err
|
||
}
|
||
return affected, nil
|
||
}
|
||
|
||
func statDay(ms int64) string {
|
||
return time.UnixMilli(ms).UTC().Format("2006-01-02")
|
||
}
|
||
|
||
func normalizeID(id int64) int64 {
|
||
if id < 0 {
|
||
return 0
|
||
}
|
||
return id
|
||
}
|
||
|
||
func normalizeShortText(value string) string {
|
||
return strings.TrimSpace(value)
|
||
}
|
||
|
||
func normalizeDimension(countryID int64, regionID int64) (int64, int64) {
|
||
return normalizeID(countryID), normalizeID(regionID)
|
||
}
|
||
|
||
func DecodeJSON(payload string) map[string]any {
|
||
decoded := map[string]any{}
|
||
_ = json.Unmarshal([]byte(payload), &decoded)
|
||
return decoded
|
||
}
|
||
|
||
func Int64(payload map[string]any, key string) int64 {
|
||
switch value := payload[key].(type) {
|
||
case float64:
|
||
return int64(value)
|
||
case int64:
|
||
return value
|
||
case json.Number:
|
||
v, _ := value.Int64()
|
||
return v
|
||
default:
|
||
return 0
|
||
}
|
||
}
|
||
|
||
func String(payload map[string]any, key string) string {
|
||
if value, ok := payload[key].(string); ok {
|
||
return strings.TrimSpace(value)
|
||
}
|
||
return ""
|
||
}
|