hy/hyapp-server
2
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
hyapp-server/services/statistics-service/internal/storage/mysql/repository.go
zhx ccc2a8b108 子币商以及充值流水
2026-07-09 18:15:57 +08:00

2384 lines
104 KiB
Go
Raw Blame History

This file contains ambiguous Unicode characters

This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.

package mysql
import (
"bytes"
"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"
SourceFinance = "finance"
)
const (
appTrackingMaxBatchSize = 50
appTrackingMaxPropertiesBytes = 4096
)
// 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
}
// database/sql defaults to unlimited connections; cap the pool so bursts
// cannot exhaust MySQL max_connections.
db.SetMaxOpenConns(20)
db.SetMaxIdleConns(10)
db.SetConnMaxLifetime(30 * time.Minute)
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
}
// The activity connection only serves low-volume snapshot reads, so it
// gets a smaller cap than the primary pool.
activityDB.SetMaxOpenConns(8)
activityDB.SetMaxIdleConns(4)
activityDB.SetConnMaxLifetime(30 * time.Minute)
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_event_consumption_type (app_code, source, event_type, consumed_at_ms)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci`,
`CREATE TABLE IF NOT EXISTS app_tracking_events (
app_code VARCHAR(32) NOT NULL, event_id VARCHAR(160) NOT NULL, event_name VARCHAR(96) NOT NULL,
event_type VARCHAR(64) NOT NULL DEFAULT '', screen VARCHAR(128) NOT NULL DEFAULT '',
target_type VARCHAR(64) NOT NULL DEFAULT '', target_id VARCHAR(128) NOT NULL DEFAULT '',
user_id BIGINT NOT NULL DEFAULT 0, device_id VARCHAR(128) NOT NULL DEFAULT '',
session_id VARCHAR(160) NOT NULL DEFAULT '', platform VARCHAR(32) NOT NULL DEFAULT '',
app_version VARCHAR(64) NOT NULL DEFAULT '', language VARCHAR(32) NOT NULL DEFAULT '',
timezone VARCHAR(64) NOT NULL DEFAULT '', country_id BIGINT NOT NULL DEFAULT 0,
region_id BIGINT NOT NULL DEFAULT 0, duration_ms BIGINT NOT NULL DEFAULT 0,
success TINYINT(1) NOT NULL DEFAULT 0, error_code VARCHAR(64) NOT NULL DEFAULT '',
properties_json JSON NOT NULL, occurred_at_ms BIGINT NOT NULL, received_at_ms BIGINT NOT NULL,
stat_day DATE NOT NULL,
PRIMARY KEY (app_code, event_id),
KEY idx_app_tracking_event_name (app_code, stat_day, event_name),
KEY idx_app_tracking_screen (app_code, stat_day, screen),
KEY idx_app_tracking_user (app_code, user_id, occurred_at_ms),
KEY idx_app_tracking_device (app_code, device_id, occurred_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_tz VARCHAR(64) NOT NULL DEFAULT 'UTC', 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, coin_seller_stock_coin 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,
coin_total BIGINT NOT NULL DEFAULT 0, consumed_coin BIGINT NOT NULL DEFAULT 0,
output_coin BIGINT NOT NULL DEFAULT 0, platform_grant_coin BIGINT NOT NULL DEFAULT 0, manual_grant_coin BIGINT NOT NULL DEFAULT 0,
salary_usd_minor BIGINT NOT NULL DEFAULT 0, salary_transfer_coin BIGINT NOT NULL DEFAULT 0,
mic_online_ms BIGINT NOT NULL DEFAULT 0, mic_online_users 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_tz, stat_day, country_id, region_id),
KEY idx_stat_app_day_region (app_code, stat_tz, 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_tz VARCHAR(64) NOT NULL DEFAULT 'UTC', 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_tz, stat_day, user_id),
KEY idx_stat_user_day_country (app_code, stat_tz, stat_day, country_id),
KEY idx_stat_user_day_region (app_code, stat_tz, 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, stat_tz VARCHAR(64) NOT NULL DEFAULT 'UTC', 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, stat_tz, user_id), KEY idx_stat_user_registration_day (app_code, stat_tz, registered_day, country_id),
KEY idx_stat_user_registration_region (app_code, stat_tz, registered_day, region_id),
KEY idx_stat_user_registration_country_day (app_code, stat_tz, country_id, registered_day, user_id),
KEY idx_stat_user_registration_region_day (app_code, stat_tz, region_id, registered_day, country_id, user_id)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci`,
`CREATE TABLE IF NOT EXISTS stat_user_dimension (
app_code VARCHAR(32) NOT NULL, user_id BIGINT NOT NULL,
country_id BIGINT NOT NULL DEFAULT 0, region_id BIGINT NOT NULL DEFAULT 0,
updated_at_ms BIGINT NOT NULL,
PRIMARY KEY (app_code, user_id),
KEY idx_stat_user_dimension_country (app_code, country_id),
KEY idx_stat_user_dimension_region (app_code, region_id)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci`,
`CREATE TABLE IF NOT EXISTS stat_user_day_mic (
app_code VARCHAR(32) NOT NULL, stat_tz VARCHAR(64) NOT NULL DEFAULT 'UTC', stat_day DATE NOT NULL,
country_id BIGINT NOT NULL DEFAULT 0, region_id BIGINT NOT NULL DEFAULT 0,
user_id BIGINT NOT NULL, mic_online_ms BIGINT NOT NULL DEFAULT 0, updated_at_ms BIGINT NOT NULL,
PRIMARY KEY (app_code, stat_tz, stat_day, user_id),
KEY idx_stat_user_day_mic_country (app_code, stat_tz, stat_day, country_id),
KEY idx_stat_user_day_mic_region (app_code, stat_tz, stat_day, region_id)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci`,
`CREATE TABLE IF NOT EXISTS stat_social_user_day (
app_code VARCHAR(32) NOT NULL, stat_tz VARCHAR(64) NOT NULL DEFAULT 'UTC', stat_day DATE NOT NULL,
subject_key VARCHAR(160) NOT NULL, user_id BIGINT NOT NULL DEFAULT 0, device_id VARCHAR(128) NOT NULL DEFAULT '',
country_id BIGINT NOT NULL DEFAULT 0, region_id BIGINT NOT NULL DEFAULT 0, user_role VARCHAR(16) NOT NULL DEFAULT 'user',
app_first_open TINYINT(1) NOT NULL DEFAULT 0, registered_success TINYINT(1) NOT NULL DEFAULT 0,
active_user TINYINT(1) NOT NULL DEFAULT 0, room_join_success TINYINT(1) NOT NULL DEFAULT 0, room_join_fail TINYINT(1) NOT NULL DEFAULT 0,
mic_up_success TINYINT(1) NOT NULL DEFAULT 0, gift_sent TINYINT(1) NOT NULL DEFAULT 0, normal_gift_sent TINYINT(1) NOT NULL DEFAULT 0,
lucky_gift_sent TINYINT(1) NOT NULL DEFAULT 0, super_lucky_gift_sent TINYINT(1) NOT NULL DEFAULT 0,
recharge_user TINYINT(1) NOT NULL DEFAULT 0, first_recharge_user TINYINT(1) NOT NULL DEFAULT 0, repeat_recharge_user TINYINT(1) NOT NULL DEFAULT 0,
game_bet_user TINYINT(1) NOT NULL DEFAULT 0, self_game_bet_user TINYINT(1) NOT NULL DEFAULT 0, probability_game_bet_user TINYINT(1) NOT NULL DEFAULT 0,
chat_sent TINYINT(1) NOT NULL DEFAULT 0, follow_user TINYINT(1) NOT NULL DEFAULT 0, follow_room TINYINT(1) NOT NULL DEFAULT 0,
gift_panel_open TINYINT(1) NOT NULL DEFAULT 0, probability_game_panel_open TINYINT(1) NOT NULL DEFAULT 0,
app_stay_ms BIGINT NOT NULL DEFAULT 0, room_stay_ms BIGINT NOT NULL DEFAULT 0, mic_stay_ms BIGINT NOT NULL DEFAULT 0,
recharge_count BIGINT NOT NULL DEFAULT 0, recharge_usd_minor BIGINT NOT NULL DEFAULT 0,
google_recharge_usd_minor BIGINT NOT NULL DEFAULT 0, third_party_recharge_usd_minor BIGINT NOT NULL DEFAULT 0, coin_seller_recharge_usd_minor BIGINT NOT NULL DEFAULT 0,
first_recharge_usd_minor BIGINT NOT NULL DEFAULT 0, repeat_recharge_usd_minor BIGINT NOT NULL DEFAULT 0,
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,
lucky_gift_payout_coin BIGINT NOT NULL DEFAULT 0, super_lucky_gift_payout_coin BIGINT NOT NULL DEFAULT 0,
game_bet_coin BIGINT NOT NULL DEFAULT 0, game_bet_count BIGINT NOT NULL DEFAULT 0,
self_game_bet_coin BIGINT NOT NULL DEFAULT 0, self_game_bet_count BIGINT NOT NULL DEFAULT 0,
probability_game_bet_coin BIGINT NOT NULL DEFAULT 0, probability_game_bet_count BIGINT NOT NULL DEFAULT 0,
other_consume_coin BIGINT NOT NULL DEFAULT 0, output_coin BIGINT NOT NULL DEFAULT 0, updated_at_ms BIGINT NOT NULL,
PRIMARY KEY (app_code, stat_tz, stat_day, subject_key),
KEY idx_stat_social_user_day_user (app_code, stat_tz, stat_day, user_id),
KEY idx_stat_social_user_day_region (app_code, stat_tz, stat_day, region_id),
KEY idx_stat_social_user_day_filter (app_code, stat_tz, stat_day, user_role, recharge_user)
) 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_tz VARCHAR(64) NOT NULL DEFAULT 'UTC', 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_tz, stat_day, user_id), KEY idx_stat_recharge_payer_country (app_code, stat_tz, stat_day, country_id),
KEY idx_stat_recharge_payer_region (app_code, stat_tz, 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_tz VARCHAR(64) NOT NULL DEFAULT 'UTC', 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_tz, stat_day, user_id),
KEY idx_stat_lucky_payer_country (app_code, stat_tz, stat_day, country_id),
KEY idx_stat_lucky_payer_region (app_code, stat_tz, stat_day, region_id)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci`,
`CREATE TABLE IF NOT EXISTS stat_platform_grant_coin_events (
app_code VARCHAR(32) NOT NULL, stat_tz VARCHAR(64) NOT NULL DEFAULT 'UTC', stat_day DATE NOT NULL,
event_id VARCHAR(160) NOT NULL, event_type VARCHAR(64) NOT NULL,
user_id BIGINT NOT NULL, country_id BIGINT NOT NULL DEFAULT 0, region_id BIGINT NOT NULL DEFAULT 0,
amount BIGINT NOT NULL DEFAULT 0, occurred_at_ms BIGINT NOT NULL, updated_at_ms BIGINT NOT NULL,
PRIMARY KEY (app_code, stat_tz, event_id),
KEY idx_stat_platform_grant_country (app_code, stat_tz, stat_day, country_id, region_id, amount),
KEY idx_stat_platform_grant_user (app_code, stat_tz, stat_day, country_id, user_id, occurred_at_ms),
KEY idx_stat_platform_grant_source (app_code, stat_tz, stat_day, event_type, country_id, user_id, occurred_at_ms)
) 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_tz VARCHAR(64) NOT NULL DEFAULT 'UTC', 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_tz, stat_day, country_id, region_id, pool_id),
KEY idx_stat_lucky_pool_overview (app_code, stat_tz, stat_day, pool_id),
KEY idx_stat_lucky_pool_region (app_code, stat_tz, 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_tz VARCHAR(64) NOT NULL DEFAULT 'UTC', 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_tz, stat_day, country_id, region_id, room_id),
KEY idx_stat_real_room_robot_gift_region (app_code, stat_tz, stat_day, region_id),
KEY idx_stat_real_room_robot_gift_country (app_code, stat_tz, 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_tz VARCHAR(64) NOT NULL DEFAULT 'UTC', 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_tz, stat_day, country_id, region_id, platform_code, game_id),
KEY idx_stat_game_day_region (app_code, stat_tz, 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_tz VARCHAR(64) NOT NULL DEFAULT 'UTC', 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_tz, stat_day, country_id, platform_code, game_id, user_id),
KEY idx_stat_game_player_region (app_code, stat_tz, 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_tz VARCHAR(64) NOT NULL DEFAULT 'UTC', 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_tz, stat_day, country_id, region_id, game_id, event_name, language, client_version, h5_version, entry_source),
KEY idx_stat_self_game_h5_event_name (app_code, stat_tz, stat_day, game_id, event_name),
KEY idx_stat_self_game_h5_day (app_code, stat_tz, 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_tz VARCHAR(64) NOT NULL DEFAULT 'UTC', 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_tz, stat_day, game_id, event_name, user_id),
KEY idx_stat_self_game_h5_users_day (app_code, stat_tz, 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_tz VARCHAR(64) NOT NULL DEFAULT 'UTC', 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_tz, stat_day, country_id, region_id, game_id, stake_coin),
KEY idx_stat_self_game_match_game (app_code, stat_tz, stat_day, game_id),
KEY idx_stat_self_game_match_region (app_code, stat_tz, 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_tz VARCHAR(64) NOT NULL DEFAULT 'UTC', 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_tz, stat_day, game_id, direction, reason),
KEY idx_stat_self_game_pool_game (app_code, stat_tz, 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_tz VARCHAR(64) NOT NULL DEFAULT 'UTC', 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_tz, stat_day, country_id, region_id, game_id, stake_coin, participant_type, result, rps_gesture, dice_point),
KEY idx_stat_self_game_participant_game (app_code, stat_tz, stat_day, game_id),
KEY idx_stat_self_game_participant_region (app_code, stat_tz, 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_tz VARCHAR(64) NOT NULL DEFAULT 'UTC', 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_tz, stat_day, game_id, user_id),
KEY idx_stat_self_game_user_win (app_code, stat_tz, stat_day, game_id, net_win_coin),
KEY idx_stat_self_game_user_region (app_code, stat_tz, 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
}
}
for _, table := range []string{
"stat_app_day_country",
"stat_user_day_activity",
"stat_user_registration",
"stat_user_day_mic",
"stat_social_user_day",
"stat_recharge_day_payers",
"stat_lucky_gift_day_payers",
"stat_lucky_gift_pool_day_country",
"stat_real_room_robot_gift_day_rooms",
"stat_game_day_country",
"stat_game_day_players",
"stat_self_game_h5_event_day",
"stat_self_game_h5_event_users",
"stat_self_game_match_day",
"stat_self_game_pool_day",
"stat_self_game_participant_day",
"stat_self_game_user_day",
} {
// stat_tz 默认补成 UTC，旧表升级后历史行仍保持原 UTC 口径；新增的 Asia/Shanghai 行由实时消费和回灌任务单独写入。
if _, err := r.db.ExecContext(ctx, "ALTER TABLE "+table+" ADD COLUMN stat_tz VARCHAR(64) NOT NULL DEFAULT 'UTC' AFTER app_code"); err != nil && !isDuplicateSchemaError(err) {
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_app_day_country ADD COLUMN coin_seller_stock_coin BIGINT NOT NULL DEFAULT 0 AFTER coin_seller_recharge_usd_minor`,
`ALTER TABLE stat_app_day_country ADD COLUMN coin_total BIGINT NOT NULL DEFAULT 0 AFTER coin_seller_transfer_coin`,
`ALTER TABLE stat_app_day_country ADD COLUMN consumed_coin BIGINT NOT NULL DEFAULT 0 AFTER coin_total`,
`ALTER TABLE stat_app_day_country ADD COLUMN output_coin BIGINT NOT NULL DEFAULT 0 AFTER consumed_coin`,
`ALTER TABLE stat_app_day_country ADD COLUMN platform_grant_coin BIGINT NOT NULL DEFAULT 0 AFTER output_coin`,
`ALTER TABLE stat_app_day_country ADD COLUMN manual_grant_coin BIGINT NOT NULL DEFAULT 0 AFTER platform_grant_coin`,
`ALTER TABLE stat_app_day_country ADD COLUMN salary_usd_minor BIGINT NOT NULL DEFAULT 0 AFTER manual_grant_coin`,
`ALTER TABLE stat_app_day_country ADD COLUMN salary_transfer_coin BIGINT NOT NULL DEFAULT 0 AFTER salary_usd_minor`,
`ALTER TABLE stat_app_day_country ADD COLUMN mic_online_ms BIGINT NOT NULL DEFAULT 0 AFTER salary_transfer_coin`,
`ALTER TABLE stat_app_day_country ADD COLUMN mic_online_users BIGINT NOT NULL DEFAULT 0 AFTER mic_online_ms`,
`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_user_day_mic 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_tz, stat_day, region_id)`,
`ALTER TABLE stat_user_day_activity ADD KEY idx_stat_user_day_region (app_code, stat_tz, stat_day, region_id)`,
`ALTER TABLE stat_user_registration ADD KEY idx_stat_user_registration_region (app_code, stat_tz, registered_day, region_id)`,
`ALTER TABLE stat_user_day_mic ADD KEY idx_stat_user_day_mic_region (app_code, stat_tz, stat_day, region_id)`,
`ALTER TABLE stat_recharge_day_payers ADD KEY idx_stat_recharge_payer_region (app_code, stat_tz, stat_day, region_id)`,
`ALTER TABLE stat_lucky_gift_day_payers ADD KEY idx_stat_lucky_payer_region (app_code, stat_tz, stat_day, region_id)`,
`ALTER TABLE stat_lucky_gift_pool_day_country ADD KEY idx_stat_lucky_pool_region (app_code, stat_tz, stat_day, region_id)`,
`ALTER TABLE stat_game_day_country ADD KEY idx_stat_game_day_region (app_code, stat_tz, stat_day, region_id)`,
`ALTER TABLE stat_game_day_players ADD KEY idx_stat_game_player_region (app_code, stat_tz, 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{
"app_tracking_events": {"app_code", "event_id"},
"stat_app_day_country": {"app_code", "stat_tz", "stat_day", "country_id", "region_id"},
"stat_user_day_activity": {"app_code", "stat_tz", "stat_day", "user_id"},
"stat_user_registration": {"app_code", "stat_tz", "user_id"},
"stat_user_dimension": {"app_code", "user_id"},
"stat_user_day_mic": {"app_code", "stat_tz", "stat_day", "user_id"},
"stat_social_user_day": {"app_code", "stat_tz", "stat_day", "subject_key"},
"stat_recharge_day_payers": {"app_code", "stat_tz", "stat_day", "user_id"},
"stat_lucky_gift_day_payers": {"app_code", "stat_tz", "stat_day", "user_id"},
"stat_platform_grant_coin_events": {"app_code", "stat_tz", "event_id"},
"stat_lucky_gift_pool_day_country": {"app_code", "stat_tz", "stat_day", "country_id", "region_id", "pool_id"},
"stat_real_room_robot_gift_day_rooms": {"app_code", "stat_tz", "stat_day", "country_id", "region_id", "room_id"},
"stat_game_day_country": {"app_code", "stat_tz", "stat_day", "country_id", "region_id", "platform_code", "game_id"},
"stat_game_day_players": {"app_code", "stat_tz", "stat_day", "country_id", "platform_code", "game_id", "user_id"},
"stat_self_game_h5_event_day": {"app_code", "stat_tz", "stat_day", "country_id", "region_id", "game_id", "event_name", "language", "client_version", "h5_version", "entry_source"},
"stat_self_game_h5_event_users": {"app_code", "stat_tz", "stat_day", "game_id", "event_name", "user_id"},
"stat_self_game_match_day": {"app_code", "stat_tz", "stat_day", "country_id", "region_id", "game_id", "stake_coin"},
"stat_self_game_pool_day": {"app_code", "stat_tz", "stat_day", "game_id", "direction", "reason"},
"stat_self_game_participant_day": {"app_code", "stat_tz", "stat_day", "country_id", "region_id", "game_id", "stake_coin", "participant_type", "result", "rps_gesture", "dice_point"},
"stat_self_game_user_day": {"app_code", "stat_tz", "stat_day", "game_id", "user_id"},
} {
if err := r.ensurePrimaryKeyColumns(ctx, table, columns); err != nil {
return err
}
}
for _, index := range []struct {
table string
name string
columns []string
}{
{"app_tracking_events", "idx_app_tracking_event_name", []string{"app_code", "stat_day", "event_name"}},
{"app_tracking_events", "idx_app_tracking_screen", []string{"app_code", "stat_day", "screen"}},
{"app_tracking_events", "idx_app_tracking_user", []string{"app_code", "user_id", "occurred_at_ms"}},
{"app_tracking_events", "idx_app_tracking_device", []string{"app_code", "device_id", "occurred_at_ms"}},
{"stat_app_day_country", "idx_stat_app_day_region", []string{"app_code", "stat_tz", "stat_day", "region_id"}},
{"stat_user_day_activity", "idx_stat_user_day_country", []string{"app_code", "stat_tz", "stat_day", "country_id"}},
{"stat_user_day_activity", "idx_stat_user_day_region", []string{"app_code", "stat_tz", "stat_day", "region_id"}},
{"stat_user_registration", "idx_stat_user_registration_day", []string{"app_code", "stat_tz", "registered_day", "country_id"}},
{"stat_user_registration", "idx_stat_user_registration_region", []string{"app_code", "stat_tz", "registered_day", "region_id"}},
{"stat_user_dimension", "idx_stat_user_dimension_country", []string{"app_code", "country_id"}},
{"stat_user_dimension", "idx_stat_user_dimension_region", []string{"app_code", "region_id"}},
{"stat_user_day_mic", "idx_stat_user_day_mic_country", []string{"app_code", "stat_tz", "stat_day", "country_id"}},
{"stat_user_day_mic", "idx_stat_user_day_mic_region", []string{"app_code", "stat_tz", "stat_day", "region_id"}},
{"stat_social_user_day", "idx_stat_social_user_day_user", []string{"app_code", "stat_tz", "stat_day", "user_id"}},
{"stat_social_user_day", "idx_stat_social_user_day_region", []string{"app_code", "stat_tz", "stat_day", "region_id"}},
{"stat_social_user_day", "idx_stat_social_user_day_filter", []string{"app_code", "stat_tz", "stat_day", "user_role", "recharge_user"}},
{"stat_recharge_day_payers", "idx_stat_recharge_payer_country", []string{"app_code", "stat_tz", "stat_day", "country_id"}},
{"stat_recharge_day_payers", "idx_stat_recharge_payer_region", []string{"app_code", "stat_tz", "stat_day", "region_id"}},
{"stat_lucky_gift_day_payers", "idx_stat_lucky_payer_country", []string{"app_code", "stat_tz", "stat_day", "country_id"}},
{"stat_lucky_gift_day_payers", "idx_stat_lucky_payer_region", []string{"app_code", "stat_tz", "stat_day", "region_id"}},
{"stat_platform_grant_coin_events", "idx_stat_platform_grant_country", []string{"app_code", "stat_tz", "stat_day", "country_id", "region_id", "amount"}},
{"stat_platform_grant_coin_events", "idx_stat_platform_grant_user", []string{"app_code", "stat_tz", "stat_day", "country_id", "user_id", "occurred_at_ms"}},
{"stat_platform_grant_coin_events", "idx_stat_platform_grant_source", []string{"app_code", "stat_tz", "stat_day", "event_type", "country_id", "user_id", "occurred_at_ms"}},
{"stat_lucky_gift_pool_day_country", "idx_stat_lucky_pool_overview", []string{"app_code", "stat_tz", "stat_day", "pool_id"}},
{"stat_lucky_gift_pool_day_country", "idx_stat_lucky_pool_region", []string{"app_code", "stat_tz", "stat_day", "region_id"}},
{"stat_real_room_robot_gift_day_rooms", "idx_stat_real_room_robot_gift_region", []string{"app_code", "stat_tz", "stat_day", "region_id"}},
{"stat_real_room_robot_gift_day_rooms", "idx_stat_real_room_robot_gift_country", []string{"app_code", "stat_tz", "stat_day", "country_id"}},
{"stat_game_day_country", "idx_stat_game_day_region", []string{"app_code", "stat_tz", "stat_day", "region_id"}},
{"stat_game_day_players", "idx_stat_game_player_region", []string{"app_code", "stat_tz", "stat_day", "region_id"}},
{"stat_self_game_h5_event_day", "idx_stat_self_game_h5_event_name", []string{"app_code", "stat_tz", "stat_day", "game_id", "event_name"}},
{"stat_self_game_h5_event_day", "idx_stat_self_game_h5_day", []string{"app_code", "stat_tz", "stat_day", "region_id", "game_id"}},
{"stat_self_game_h5_event_users", "idx_stat_self_game_h5_users_day", []string{"app_code", "stat_tz", "stat_day", "game_id", "event_name"}},
{"stat_self_game_match_day", "idx_stat_self_game_match_game", []string{"app_code", "stat_tz", "stat_day", "game_id"}},
{"stat_self_game_match_day", "idx_stat_self_game_match_region", []string{"app_code", "stat_tz", "stat_day", "region_id", "game_id"}},
{"stat_self_game_pool_day", "idx_stat_self_game_pool_game", []string{"app_code", "stat_tz", "stat_day", "game_id"}},
{"stat_self_game_participant_day", "idx_stat_self_game_participant_game", []string{"app_code", "stat_tz", "stat_day", "game_id"}},
{"stat_self_game_participant_day", "idx_stat_self_game_participant_region", []string{"app_code", "stat_tz", "stat_day", "region_id", "game_id"}},
{"stat_self_game_user_day", "idx_stat_self_game_user_win", []string{"app_code", "stat_tz", "stat_day", "game_id", "net_win_coin"}},
{"stat_self_game_user_day", "idx_stat_self_game_user_region", []string{"app_code", "stat_tz", "stat_day", "region_id", "game_id"}},
} {
if err := r.ensureIndexColumns(ctx, index.table, index.name, index.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 (r *Repository) ensureIndexColumns(ctx context.Context, table string, index string, expected []string) error {
rows, err := r.db.QueryContext(ctx, `
SELECT COLUMN_NAME
FROM information_schema.STATISTICS
WHERE TABLE_SCHEMA = DATABASE() AND TABLE_NAME = ? AND INDEX_NAME = ?
ORDER BY SEQ_IN_INDEX`, table, index)
if err != nil {
return err
}
defer rows.Close()
actual := []string{}
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 strings.Join(actual, ",") == strings.Join(expected, ",") {
return nil
}
// 老库里多数二级索引已经同名存在，但列序缺少 stat_tz；这里按信息_schema 的实际结构修复，保证北京日查询不会退化成跨时区扫描。
if len(actual) > 0 {
if _, err := r.db.ExecContext(ctx, "ALTER TABLE "+table+" DROP INDEX "+index); err != nil && !isMissingIndexError(err) {
return err
}
}
_, err = r.db.ExecContext(ctx, "ALTER TABLE "+table+" ADD KEY "+index+" ("+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
Role string
CountryID int64
RegionID int64
OccurredAtMS int64
}
type UserRegionChangedEvent 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 FinanceCoinSellerRechargeOrderEvent struct {
AppCode string
EventID string
VerifyStatus string
TargetUserID int64
USDMinor int64
CoinAmount int64
CountryID int64
RegionID int64
CreatedAtMS 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 PlatformGrantCoinEvent struct {
AppCode string
EventID string
EventType string
UserID int64
CountryID int64
RegionID int64
Amount int64
OccurredAtMS int64
}
type CoinOutputEvent struct {
AppCode string
EventID string
EventType string
UserID int64
CountryID int64
RegionID int64
Amount int64
OccurredAtMS int64
}
type WalletBalanceChangedEvent struct {
AppCode string
EventID string
EventType string
UserID int64
DimensionUserID int64
AssetType string
BizType string
CommandID string
AvailableDelta int64
FrozenDelta int64
CountryID int64
RegionID int64
OccurredAtMS int64
}
type UserMicDailyStatsUpdatedEvent struct {
AppCode string
EventID string
UserID int64
StatDate string
MicOnlineDeltaMS int64
OccurredAtMS int64
}
type UserActiveEvent struct {
AppCode string
EventID string
EventType string
UserID int64
Role string
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 AppTrackingEvent struct {
AppCode string
EventID string
EventName string
EventType string
Screen string
TargetType string
TargetID string
UserID int64
DeviceID string
SessionID string
Platform string
AppVersion string
Language string
Timezone string
CountryID int64
RegionID int64
DurationMS int64
Success bool
ErrorCode string
Properties json.RawMessage
OccurredAtMS int64
}
type AppTrackingReportResult struct {
Accepted bool `json:"accepted"`
Received int `json:"received"`
Stored int `json:"stored"`
Duplicated int `json:"duplicated"`
ServerTimeMS int64 `json:"server_time_ms"`
}
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 {
countryID, regionID := normalizeDimension(event.CountryID, event.RegionID)
if err := upsertUserDimension(ctx, tx, event.AppCode, event.UserID, countryID, regionID, nowMS); err != nil {
return err
}
for _, scope := range statDayScopesFromContext(ctx, event.OccurredAtMS) {
if err := ensureAppDay(ctx, tx, event.AppCode, scope.tz, scope.day, countryID, regionID, nowMS); err != nil {
return err
}
inserted, err := insertUnique(ctx, tx, `
INSERT IGNORE INTO stat_user_registration (app_code, stat_tz, user_id, registered_day, country_id, region_id, registered_at_ms)
VALUES (?, ?, ?, ?, ?, ?, ?)
`, appcode.Normalize(event.AppCode), scope.tz, event.UserID, scope.day, countryID, regionID, event.OccurredAtMS)
if err != nil {
return err
}
// MQ 重投和补偿任务可能带来新的 event_id，但同一用户在同一统计时区的注册 cohort 只能入库一次；
// new_users 必须按 INSERT IGNORE 的真实插入行数累加，避免回放把国家新增人数重复放大。
if _, err = tx.ExecContext(ctx, `
UPDATE stat_app_day_country
SET new_users = new_users + ?, updated_at_ms = ?
WHERE app_code = ? AND stat_tz = ? AND stat_day = ? AND country_id = ? AND region_id = ?
`, inserted, nowMS, appcode.Normalize(event.AppCode), scope.tz, scope.day, countryID, regionID); err != nil {
return err
}
if inserted == 0 {
continue
}
// 产品口径要求“完成注册即产生当日活跃”：注册完成本身已经证明用户打开并完成 App 主流程；
// 这里复用活跃唯一表按 stat_tz+user+day 收敛，UTC 和中国日互不覆盖。
if err := applyActive(ctx, tx, event.AppCode, scope.tz, scope.day, countryID, regionID, event.UserID, event.OccurredAtMS, nowMS); err != nil {
return err
}
if err := applySocialUserDayDelta(ctx, tx, socialUserDayDelta{
AppCode: event.AppCode,
StatTZ: scope.tz,
StatDay: scope.day,
UserID: event.UserID,
CountryID: countryID,
RegionID: regionID,
UserRole: event.Role,
UpdatedAt: nowMS,
Registered: 1,
Active: 1,
}); err != nil {
return err
}
}
return nil
})
}
func (r *Repository) ConsumeUserRegionChanged(ctx context.Context, event UserRegionChangedEvent) error {
return r.withEvent(ctx, SourceUser, event.EventID, "UserRegionChanged", func(tx *sql.Tx, nowMS int64) error {
countryID, regionID := normalizeDimension(event.CountryID, event.RegionID)
return upsertUserDimension(ctx, tx, event.AppCode, event.UserID, countryID, regionID, nowMS)
})
}
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 {
countryID, regionID := normalizeDimension(event.TargetCountryID, event.TargetRegionID)
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", "v5pay":
mifapayUSD = event.USDMinor
case "usdt_trc20":
// USDT-TRC20 是站外 H5 充值渠道；当前大盘只单列 MiFaPay/Google/币商转账，
// 因此只进入总充值和付费用户，不误归到币商美元充值桶。
case "coin_seller_transfer", "coin_seller":
coinSellerTransferCoin = event.CoinAmount
default:
coinSellerUSD = event.USDMinor
}
for _, scope := range statDayScopesFromContext(ctx, event.OccurredAtMS) {
if err := ensureAppDay(ctx, tx, event.AppCode, scope.tz, scope.day, countryID, regionID, nowMS); err != nil {
return err
}
paidUsers, err := insertUnique(ctx, tx, `
INSERT IGNORE INTO stat_recharge_day_payers (app_code, stat_tz, stat_day, country_id, region_id, user_id, first_paid_at_ms)
VALUES (?, ?, ?, ?, ?, ?, ?)
`, appcode.Normalize(event.AppCode), scope.tz, scope.day, countryID, regionID, event.UserID, event.OccurredAtMS)
if err != nil {
return err
}
if _, 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_tz = ? AND stat_day = ? AND country_id = ? AND region_id = ?
`, event.USDMinor, newUserUSD, coinSellerUSD, mifapayUSD, googleUSD, coinSellerTransferCoin, paidUsers, nowMS, appcode.Normalize(event.AppCode), scope.tz, scope.day, countryID, regionID); err != nil {
return err
}
googleUSD, thirdPartyUSD, coinSellerUSD := int64(0), int64(0), int64(0)
switch strings.ToLower(strings.TrimSpace(event.RechargeType)) {
case "google", "google_play":
googleUSD = event.USDMinor
case "coin_seller_transfer", "coin_seller":
coinSellerUSD = event.USDMinor
}
firstRecharge, repeatRecharge, firstUSD, repeatUSD := int64(0), int64(0), int64(0), int64(0)
if event.RechargeSequence == 1 {
firstRecharge, firstUSD = 1, event.USDMinor
} else if event.RechargeSequence > 1 {
repeatRecharge, repeatUSD = 1, event.USDMinor
}
if err := applySocialUserDayDelta(ctx, tx, socialUserDayDelta{
AppCode: event.AppCode,
StatTZ: scope.tz,
StatDay: scope.day,
UserID: event.UserID,
CountryID: countryID,
RegionID: regionID,
UpdatedAt: nowMS,
Recharge: 1,
FirstRecharge: firstRecharge,
RepeatRecharge: repeatRecharge,
RechargeCount: 1,
RechargeUSDMinor: event.USDMinor,
GoogleRechargeUSDMinor: googleUSD,
ThirdPartyRechargeUSDMinor: thirdPartyUSD,
CoinSellerRechargeUSDMinor: coinSellerUSD,
FirstRechargeUSDMinor: firstUSD,
RepeatRechargeUSDMinor: repeatUSD,
}); err != nil {
return err
}
}
return nil
})
}
func (r *Repository) ConsumeFinanceCoinSellerRechargeOrder(ctx context.Context, event FinanceCoinSellerRechargeOrderEvent) error {
return r.withEvent(ctx, SourceFinance, event.EventID, "FinanceCoinSellerRechargeOrderVerified", func(tx *sql.Tx, nowMS int64) error {
if !strings.EqualFold(strings.TrimSpace(event.VerifyStatus), "verified") {
return nil
}
if event.TargetUserID <= 0 || event.USDMinor <= 0 {
return nil
}
occurredAtMS := event.CreatedAtMS
if occurredAtMS <= 0 {
occurredAtMS = event.OccurredAtMS
}
countryID, regionID := normalizeDimension(event.CountryID, event.RegionID)
for _, scope := range statDayScopesFromContext(ctx, occurredAtMS) {
// Social BI 的第三方充值口径绑定 finance 工作台 verified 币商订单：金额和次数进入营收，
// 但不置 recharge_user/first/repeat 标记，避免财务补单把付费率和 ARPPU 分母放大。
if err := applySocialUserDayDelta(ctx, tx, socialUserDayDelta{
AppCode: event.AppCode,
StatTZ: scope.tz,
StatDay: scope.day,
UserID: event.TargetUserID,
CountryID: countryID,
RegionID: regionID,
UpdatedAt: nowMS,
RechargeCount: 1,
RechargeUSDMinor: event.USDMinor,
ThirdPartyRechargeUSDMinor: event.USDMinor,
CoinSellerRechargeUSDMinor: event.USDMinor,
}); err != nil {
return err
}
}
return nil
})
}
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 {
countryID, regionID := normalizeDimension(event.CountryID, event.RegionID)
for _, scope := range statDayScopesFromContext(ctx, event.OccurredAtMS) {
if err := ensureAppDay(ctx, tx, event.AppCode, scope.tz, scope.day, countryID, regionID, nowMS); err != nil {
return err
}
// 币商进货是后台 USDT 入金事实，只进入总充值和币商充值列；它不是普通用户到账，所以不增加用户 USDT 首充和付费用户。
if _, 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 + ?,
coin_seller_stock_coin = coin_seller_stock_coin + ?,
updated_at_ms = ?
WHERE app_code = ? AND stat_tz = ? AND stat_day = ? AND country_id = ? AND region_id = ?
`, event.USDMinor, event.USDMinor, event.CoinAmount, nowMS, appcode.Normalize(event.AppCode), scope.tz, scope.day, countryID, regionID); err != nil {
return err
}
}
return nil
})
}
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 {
countryID, regionID := normalizeDimension(event.CountryID, event.VisibleRegionID)
for _, scope := range statDayScopesFromContext(ctx, event.OccurredAtMS) {
if event.RealRoomRobot {
normalCoin, luckyCoin, superLuckyCoin := realRoomRobotGiftCoins(event)
// 真人房机器人礼物单独进 Databi 投放统计，不混入普通用户付费/幸运礼物流水；双时区行各自承接同一事实，查询时由 stat_tz 选择展示口径。
if _, err := tx.ExecContext(ctx, `
INSERT INTO stat_real_room_robot_gift_day_rooms (
app_code, stat_tz, 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), scope.tz, scope.day, countryID, regionID, strings.TrimSpace(event.RoomID), normalCoin, luckyCoin, superLuckyCoin, nowMS); err != nil {
return err
}
continue
}
if err := ensureAppDay(ctx, tx, event.AppCode, scope.tz, scope.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_tz, stat_day, country_id, region_id, user_id, first_paid_at_ms)
VALUES (?, ?, ?, ?, ?, ?, ?)
`, appcode.Normalize(event.AppCode), scope.tz, scope.day, countryID, regionID, event.SenderUserID, event.OccurredAtMS)
if err != nil {
return err
}
luckyPayers = affected
// Pool 明细是 Databi 点击下钻的真实口径：送礼和返奖都用 stat_tz+stat_day+国家+奖池对齐，避免北京时间查询误用 UTC 行。
if err := ensureLuckyGiftPoolDay(ctx, tx, event.AppCode, scope.tz, scope.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_tz = ? AND stat_day = ? AND country_id = ? AND region_id = ? AND pool_id = ?
`, luckyTurnover, nowMS, appcode.Normalize(event.AppCode), scope.tz, scope.day, countryID, regionID, poolID); err != nil {
return err
}
}
if _, 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 + ?,
consumed_coin = consumed_coin + ?,
updated_at_ms = ?
WHERE app_code = ? AND stat_tz = ? AND stat_day = ? AND country_id = ? AND region_id = ?
`, event.CoinSpent, luckyTurnover, luckyPayers, event.CoinSpent, nowMS, appcode.Normalize(event.AppCode), scope.tz, scope.day, countryID, regionID); err != nil {
return err
}
normalFlag, luckyFlag, superFlag, normalCoin, luckyCoin, superCoin := socialGiftKind(event)
if err := applySocialUserDayDelta(ctx, tx, socialUserDayDelta{
AppCode: event.AppCode,
StatTZ: scope.tz,
StatDay: scope.day,
UserID: event.SenderUserID,
CountryID: countryID,
RegionID: regionID,
UpdatedAt: nowMS,
Active: 1,
Gift: 1,
NormalGift: normalFlag,
LuckyGift: luckyFlag,
SuperGift: superFlag,
NormalGiftCoin: normalCoin,
LuckyGiftCoin: luckyCoin,
SuperGiftCoin: superCoin,
OtherConsumeCoin: 0,
}); err != nil {
return err
}
}
return nil
})
}
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 {
eventType := strings.TrimSpace(event.EventType)
if eventType == "" {
eventType = "WalletLuckyGiftRewardCredited"
}
return r.withEvent(ctx, SourceWallet, event.EventID, eventType, func(tx *sql.Tx, nowMS int64) error {
countryID, regionID := normalizeDimension(event.CountryID, event.RegionID)
amount := normalizeID(event.Amount)
poolID := strings.TrimSpace(event.PoolID)
for _, scope := range statDayScopesFromContext(ctx, event.OccurredAtMS) {
if err := ensureAppDay(ctx, tx, event.AppCode, scope.tz, scope.day, countryID, regionID, nowMS); err != nil {
return err
}
if poolID != "" {
// 钱包返奖事件带回 pool_id 后，按同一统计时区和国家写入奖池返奖；利润不落表，查询时用 turnover-payout 实时计算，避免双写漂移。
if err := ensureLuckyGiftPoolDay(ctx, tx, event.AppCode, scope.tz, scope.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_tz = ? AND stat_day = ? AND country_id = ? AND region_id = ? AND pool_id = ?
`, amount, nowMS, appcode.Normalize(event.AppCode), scope.tz, scope.day, countryID, regionID, poolID); err != nil {
return err
}
}
if _, err := tx.ExecContext(ctx, `
UPDATE stat_app_day_country
SET lucky_gift_payout = lucky_gift_payout + ?,
output_coin = output_coin + ?,
updated_at_ms = ?
WHERE app_code = ? AND stat_tz = ? AND stat_day = ? AND country_id = ? AND region_id = ?
`, amount, amount, nowMS, appcode.Normalize(event.AppCode), scope.tz, scope.day, countryID, regionID); err != nil {
return err
}
luckyPayout, superPayout := amount, int64(0)
if strings.Contains(strings.ToLower(poolID), "super_lucky") {
luckyPayout, superPayout = 0, amount
}
if err := applySocialUserDayDelta(ctx, tx, socialUserDayDelta{
AppCode: event.AppCode,
StatTZ: scope.tz,
StatDay: scope.day,
UserID: event.UserID,
CountryID: countryID,
RegionID: regionID,
UpdatedAt: nowMS,
Active: 1,
LuckyGiftPayoutCoin: luckyPayout,
SuperGiftPayoutCoin: superPayout,
OutputCoin: amount,
}); err != nil {
return err
}
}
return nil
})
}
func (r *Repository) ConsumePlatformGrantCoin(ctx context.Context, event PlatformGrantCoinEvent) error {
eventType := strings.TrimSpace(event.EventType)
if eventType == "" {
eventType = "WalletPlatformGrantCoinCredited"
}
return r.withEvent(ctx, SourceWallet, event.EventID, eventType, func(tx *sql.Tx, nowMS int64) error {
countryID, regionID, err := r.dimensionForUser(ctx, tx, event.AppCode, event.UserID, event.CountryID, event.RegionID)
if err != nil {
return err
}
amount := normalizeID(event.Amount)
if amount == 0 {
return nil
}
platformGrant, outputCoin := int64(0), int64(0)
switch {
case isOutputCoinRewardEventType(eventType):
outputCoin = amount
default:
platformGrant = amount
}
for _, scope := range statDayScopesFromContext(ctx, event.OccurredAtMS) {
if err := ensureAppDay(ctx, tx, event.AppCode, scope.tz, scope.day, countryID, regionID, nowMS); err != nil {
return err
}
if _, err := tx.ExecContext(ctx, `
UPDATE stat_app_day_country
SET platform_grant_coin = platform_grant_coin + ?,
output_coin = output_coin + ?,
updated_at_ms = ?
WHERE app_code = ? AND stat_tz = ? AND stat_day = ? AND country_id = ? AND region_id = ?
`, platformGrant, outputCoin, nowMS, appcode.Normalize(event.AppCode), scope.tz, scope.day, countryID, regionID); err != nil {
return err
}
if platformGrant > 0 {
// 保存事件级来源，后续 Databi 下钻按国家 -> 用户 -> 每条奖励来源分页查询，不需要扫 owner service 账务表。
if err := insertPlatformGrantCoinEvent(ctx, tx, event.AppCode, scope.tz, scope.day, event.EventID, eventType, event.UserID, countryID, regionID, platformGrant, event.OccurredAtMS, nowMS); err != nil {
return err
}
}
}
return nil
})
}
func (r *Repository) ConsumeCoinOutput(ctx context.Context, event CoinOutputEvent) error {
eventType := strings.TrimSpace(event.EventType)
if eventType == "" {
eventType = "WalletCoinOutputCredited"
}
return r.withEvent(ctx, SourceWallet, event.EventID, eventType, func(tx *sql.Tx, nowMS int64) error {
countryID, regionID, err := r.dimensionForUser(ctx, tx, event.AppCode, event.UserID, event.CountryID, event.RegionID)
if err != nil {
return err
}
amount := normalizeID(event.Amount)
if amount == 0 {
return nil
}
for _, scope := range statDayScopesFromContext(ctx, event.OccurredAtMS) {
if err := ensureAppDay(ctx, tx, event.AppCode, scope.tz, scope.day, countryID, regionID, nowMS); err != nil {
return err
}
// 产出金币只写统计读模型总量；来源明细仍由各 owner 的 outbox 和幂等表保留，查询层不回扫钱包流水。
if _, err := tx.ExecContext(ctx, `
UPDATE stat_app_day_country
SET output_coin = output_coin + ?, updated_at_ms = ?
WHERE app_code = ? AND stat_tz = ? AND stat_day = ? AND country_id = ? AND region_id = ?
`, amount, nowMS, appcode.Normalize(event.AppCode), scope.tz, scope.day, countryID, regionID); err != nil {
return err
}
}
return nil
})
}
func (r *Repository) ConsumeWalletBalanceChanged(ctx context.Context, event WalletBalanceChangedEvent) error {
return r.withEvent(ctx, SourceWallet, event.EventID, "WalletBalanceChanged", func(tx *sql.Tx, nowMS int64) error {
dimensionUserID := event.UserID
if event.DimensionUserID > 0 {
dimensionUserID = event.DimensionUserID
}
countryID, regionID, err := r.dimensionForUser(ctx, tx, event.AppCode, dimensionUserID, event.CountryID, event.RegionID)
if err != nil {
return err
}
totalDelta := event.AvailableDelta + event.FrozenDelta
coinTotal, consumed, outputCoin, platformGrant, manualGrant, salaryUSD, salaryTransferCoin := int64(0), int64(0), int64(0), int64(0), int64(0), int64(0), int64(0)
assetType := strings.ToUpper(strings.TrimSpace(event.AssetType))
bizType := strings.ToLower(strings.TrimSpace(event.BizType))
if assetType == "COIN" {
coinTotal = totalDelta
if totalDelta < 0 && isConsumedCoinBizType(bizType) {
consumed = -totalDelta
}
if totalDelta > 0 && isOutputCoinBizType(bizType) {
outputCoin = totalDelta
}
if totalDelta > 0 && bizType == "resource_grant" && isPlatformGrantCommandID(event.CommandID) {
platformGrant = totalDelta
}
if totalDelta > 0 && bizType == "manual_credit" {
manualGrant = totalDelta
}
}
if isHostAgencySalaryAssetType(assetType) {
// 工资总和是主播/代理工资钱包余额快照，不是结算流水。这里写 available+frozen 的净变化，
// 冻结动作本身一正一负不改变总额，入账/出账/兑换才会改变当天余额快照。
salaryUSD = totalDelta
}
if totalDelta > 0 && bizType == "salary_transfer_to_coin_seller" && assetType == "COIN_SELLER_COIN" {
salaryTransferCoin = totalDelta
}
if coinTotal == 0 && consumed == 0 && outputCoin == 0 && platformGrant == 0 && manualGrant == 0 && salaryUSD == 0 && salaryTransferCoin == 0 {
return nil
}
for _, scope := range statDayScopesFromContext(ctx, event.OccurredAtMS) {
if err := ensureAppDay(ctx, tx, event.AppCode, scope.tz, scope.day, countryID, regionID, nowMS); err != nil {
return err
}
if _, err := tx.ExecContext(ctx, `
UPDATE stat_app_day_country
SET coin_total = coin_total + ?,
consumed_coin = consumed_coin + ?,
output_coin = output_coin + ?,
platform_grant_coin = platform_grant_coin + ?,
manual_grant_coin = manual_grant_coin + ?,
salary_usd_minor = salary_usd_minor + ?,
salary_transfer_coin = salary_transfer_coin + ?,
updated_at_ms = ?
WHERE app_code = ? AND stat_tz = ? AND stat_day = ? AND country_id = ? AND region_id = ?
`, coinTotal, consumed, outputCoin, platformGrant, manualGrant, salaryUSD, salaryTransferCoin, nowMS, appcode.Normalize(event.AppCode), scope.tz, scope.day, countryID, regionID); err != nil {
return err
}
otherConsume := consumed
if bizType == "direct_gift_debit" {
otherConsume = 0
}
if otherConsume > 0 || outputCoin > 0 {
if err := applySocialUserDayDelta(ctx, tx, socialUserDayDelta{
AppCode: event.AppCode,
StatTZ: scope.tz,
StatDay: scope.day,
UserID: dimensionUserID,
CountryID: countryID,
RegionID: regionID,
UpdatedAt: nowMS,
Active: 1,
OtherConsumeCoin: otherConsume,
OutputCoin: outputCoin,
}); err != nil {
return err
}
}
if platformGrant > 0 {
// resource_grant 类入账也是平台发放金币，写入同一事件投影，确保总计、用户列表和来源明细的统计口径一致。
if err := insertPlatformGrantCoinEvent(ctx, tx, event.AppCode, scope.tz, scope.day, event.EventID, event.EventType, event.UserID, countryID, regionID, platformGrant, event.OccurredAtMS, nowMS); err != nil {
return err
}
}
}
return nil
})
}
func (r *Repository) ConsumeUserMicDailyStatsUpdated(ctx context.Context, event UserMicDailyStatsUpdatedEvent) error {
return r.withEvent(ctx, SourceUser, event.EventID, "UserMicDailyStatsUpdated", func(tx *sql.Tx, nowMS int64) error {
if event.MicOnlineDeltaMS == 0 || strings.TrimSpace(event.StatDate) == "" {
return nil
}
countryID, regionID, err := r.dimensionForUser(ctx, tx, event.AppCode, event.UserID, 0, 0)
if err != nil {
return err
}
for _, scope := range statDateScopesFromContext(ctx, event.StatDate) {
if err := ensureAppDay(ctx, tx, event.AppCode, scope.tz, scope.day, countryID, regionID, nowMS); err != nil {
return err
}
before, err := micOnlineBefore(ctx, tx, event.AppCode, scope.tz, scope.day, event.UserID)
if err != nil {
return err
}
after := before + event.MicOnlineDeltaMS
if after < 0 {
after = 0
}
userDelta := int64(0)
if before <= 0 && after > 0 {
userDelta = 1
} else if before > 0 && after <= 0 {
userDelta = -1
}
if _, err := tx.ExecContext(ctx, `
INSERT INTO stat_user_day_mic (app_code, stat_tz, stat_day, country_id, region_id, user_id, mic_online_ms, updated_at_ms)
VALUES (?, ?, ?, ?, ?, ?, ?, ?)
ON DUPLICATE KEY UPDATE
country_id = VALUES(country_id),
region_id = VALUES(region_id),
mic_online_ms = VALUES(mic_online_ms),
updated_at_ms = VALUES(updated_at_ms)
`, appcode.Normalize(event.AppCode), scope.tz, scope.day, countryID, regionID, event.UserID, after, nowMS); err != nil {
return err
}
if _, err := tx.ExecContext(ctx, `
UPDATE stat_app_day_country
SET mic_online_ms = mic_online_ms + ?,
mic_online_users = GREATEST(mic_online_users + ?, 0),
updated_at_ms = ?
WHERE app_code = ? AND stat_tz = ? AND stat_day = ? AND country_id = ? AND region_id = ?
`, after-before, userDelta, nowMS, appcode.Normalize(event.AppCode), scope.tz, scope.day, countryID, regionID); err != nil {
return err
}
micDelta := after - before
if micDelta > 0 {
if err := applySocialUserDayDelta(ctx, tx, socialUserDayDelta{
AppCode: event.AppCode,
StatTZ: scope.tz,
StatDay: scope.day,
UserID: event.UserID,
CountryID: countryID,
RegionID: regionID,
UpdatedAt: nowMS,
Active: 1,
MicUp: 1,
MicStayMS: micDelta,
}); err != nil {
return err
}
}
}
return nil
})
}
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)
for _, scope := range statDayScopesFromContext(ctx, event.OccurredAtMS) {
if err := applyActive(ctx, tx, event.AppCode, scope.tz, scope.day, countryID, regionID, event.UserID, event.OccurredAtMS, nowMS); err != nil {
return err
}
roomJoinOK := int64(0)
if strings.EqualFold(strings.TrimSpace(event.EventType), "RoomUserJoined") {
roomJoinOK = 1
}
if err := applySocialUserDayDelta(ctx, tx, socialUserDayDelta{
AppCode: event.AppCode,
StatTZ: scope.tz,
StatDay: scope.day,
UserID: event.UserID,
CountryID: countryID,
RegionID: regionID,
UserRole: event.Role,
UpdatedAt: nowMS,
Active: 1,
RoomJoinOK: roomJoinOK,
}); err != nil {
return err
}
}
return nil
})
}
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 {
countryID, regionID := normalizeDimension(event.CountryID, event.RegionID)
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
}
for _, scope := range statDayScopesFromContext(ctx, event.OccurredAtMS) {
if err := applyActive(ctx, tx, event.AppCode, scope.tz, scope.day, countryID, regionID, event.UserID, event.OccurredAtMS, nowMS); err != nil {
return err
}
if err := ensureAppDay(ctx, tx, event.AppCode, scope.tz, scope.day, countryID, regionID, nowMS); err != nil {
return err
}
playerDelta := int64(0)
if turnover > 0 {
affected, err := insertUnique(ctx, tx, `
INSERT IGNORE INTO stat_game_day_players (app_code, stat_tz, stat_day, country_id, region_id, platform_code, game_id, user_id, first_played_at_ms)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)
`, appcode.Normalize(event.AppCode), scope.tz, scope.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, scope.tz, scope.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_tz = ? AND stat_day = ? AND country_id = ? AND region_id = ? AND platform_code = ? AND game_id = ?
`, turnover, payout, refund, playerDelta, nowMS, appcode.Normalize(event.AppCode), scope.tz, scope.day, countryID, regionID, event.PlatformCode, event.GameID); err != nil {
return err
}
if _, 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 + ?,
consumed_coin = consumed_coin + ?,
output_coin = output_coin + ?,
updated_at_ms = ?
WHERE app_code = ? AND stat_tz = ? AND stat_day = ? AND country_id = ? AND region_id = ?
`, turnover, payout, refund, playerDelta, turnover, payout, nowMS, appcode.Normalize(event.AppCode), scope.tz, scope.day, countryID, regionID); err != nil {
return err
}
if turnover > 0 {
if err := applySocialUserDayDelta(ctx, tx, socialUserDayDelta{
AppCode: event.AppCode,
StatTZ: scope.tz,
StatDay: scope.day,
UserID: event.UserID,
CountryID: countryID,
RegionID: regionID,
UpdatedAt: nowMS,
Active: 1,
GameBet: 1,
ProbabilityBet: 1,
GameBetCoin: turnover,
GameBetCount: 1,
ProbabilityBetCoin: turnover,
ProbabilityBetCount: 1,
}); err != nil {
return err
}
}
}
return nil
})
}
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 {
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)
for _, scope := range statDayScopesFromContext(ctx, event.OccurredAtMS) {
if err := ensureSelfGameH5EventDay(ctx, tx, event.AppCode, scope.tz, scope.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_tz, stat_day, game_id, event_name, user_id, first_seen_at_ms)
VALUES (?, ?, ?, ?, ?, ?, ?)
`, appcode.Normalize(event.AppCode), scope.tz, scope.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 用同一 stat_tz 日、玩法和事件的用户去重，接口耗时只累加客户端实际传入的 duration。
if _, 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_tz = ? 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), scope.tz, scope.day, countryID, regionID, gameID, eventName, language, clientVersion, h5Version, entrySource); err != nil {
return err
}
}
return nil
})
}
func (r *Repository) ConsumeAppTrackingEvents(ctx context.Context, events []AppTrackingEvent) (AppTrackingReportResult, error) {
if r == nil || r.db == nil {
return AppTrackingReportResult{}, xerr.New(xerr.Unavailable, "mysql repository is not configured")
}
nowMS := time.Now().UTC().UnixMilli()
result := AppTrackingReportResult{Accepted: true, Received: len(events), ServerTimeMS: nowMS}
if len(events) == 0 || len(events) > appTrackingMaxBatchSize {
return AppTrackingReportResult{}, xerr.New(xerr.InvalidArgument, "app tracking events batch is invalid")
}
tx, err := r.db.BeginTx(ctx, nil)
if err != nil {
return AppTrackingReportResult{}, err
}
defer func() { _ = tx.Rollback() }()
for _, event := range events {
normalized, propertiesJSON, err := normalizeAppTrackingEvent(ctx, event, nowMS)
if err != nil {
return AppTrackingReportResult{}, err
}
affected, err := insertUnique(ctx, tx, `
INSERT IGNORE INTO app_tracking_events (
app_code, event_id, event_name, event_type, screen, target_type, target_id,
user_id, device_id, session_id, platform, app_version, language, timezone,
country_id, region_id, duration_ms, success, error_code, properties_json,
occurred_at_ms, received_at_ms, stat_day
) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, CAST(? AS JSON), ?, ?, ?)
`, normalized.AppCode, normalized.EventID, normalized.EventName, normalized.EventType, normalized.Screen,
normalized.TargetType, normalized.TargetID, normalized.UserID, normalized.DeviceID, normalized.SessionID,
normalized.Platform, normalized.AppVersion, normalized.Language, normalized.Timezone, normalized.CountryID,
normalized.RegionID, normalized.DurationMS, normalized.Success, normalized.ErrorCode, propertiesJSON,
normalized.OccurredAtMS, nowMS, statDay(normalized.OccurredAtMS))
if err != nil {
return AppTrackingReportResult{}, err
}
if affected > 0 {
result.Stored++
for _, scope := range statDayScopesFromContext(ctx, normalized.OccurredAtMS) {
delta, ok := socialDeltaForAppTrackingEvent(normalized, scope, nowMS)
if !ok {
continue
}
if err := applySocialUserDayDelta(ctx, tx, delta); err != nil {
return AppTrackingReportResult{}, err
}
}
} else {
result.Duplicated++
}
}
if err := tx.Commit(); err != nil {
return AppTrackingReportResult{}, err
}
return result, nil
}
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 {
countryID, regionID := normalizeDimension(event.CountryID, event.RegionID)
gameID := normalizeShortText(event.GameID)
if gameID == "" {
gameID = "unknown"
}
stakeCoin := normalizeID(event.StakeCoin)
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)
}
for _, scope := range statDayScopesFromContext(ctx, event.OccurredAtMS) {
if err := ensureSelfGameMatchDay(ctx, tx, event.AppCode, scope.tz, scope.day, countryID, regionID, gameID, stakeCoin, nowMS); err != nil {
return err
}
// 对局状态事件只按 match 级事实累加一次；明细用户维度只在结算/取消时落表，UTC 和中国日各自按 stat_tz 独立聚合。
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_tz = ? 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), scope.tz, scope.day, countryID, regionID, gameID, stakeCoin); err != nil {
return err
}
if settled > 0 {
for _, participant := range event.Participants {
if err := consumeSelfGameSettledParticipant(ctx, tx, event, participant, scope.tz, scope.day, countryID, regionID, gameID, stakeCoin, nowMS); err != nil {
return err
}
if participant.UserID <= 0 || !strings.EqualFold(strings.TrimSpace(participant.ParticipantType), "user") {
continue
}
stake := normalizeID(participant.StakeCoin)
if stake == 0 {
stake = stakeCoin
}
if err := applySocialUserDayDelta(ctx, tx, socialUserDayDelta{
AppCode: event.AppCode,
StatTZ: scope.tz,
StatDay: scope.day,
UserID: participant.UserID,
CountryID: countryID,
RegionID: regionID,
UpdatedAt: nowMS,
Active: 1,
GameBet: 1,
SelfGameBet: 1,
GameBetCoin: stake,
GameBetCount: 1,
SelfGameBetCoin: stake,
SelfGameBetCount: 1,
}); 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, scope.tz, scope.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_tz = ? AND stat_day = ? AND game_id = ? AND user_id = ?
`, nowMS, appcode.Normalize(event.AppCode), scope.tz, scope.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 {
gameID := normalizeShortText(event.GameID)
if gameID == "" {
gameID = "unknown"
}
direction := normalizeShortText(event.Direction)
reason := normalizeShortText(event.Reason)
inCoin, outCoin := int64(0), int64(0)
switch direction {
case "in":
inCoin = normalizeID(event.AmountCoin)
case "out":
outCoin = normalizeID(event.AmountCoin)
}
for _, scope := range statDayScopesFromContext(ctx, event.OccurredAtMS) {
if err := ensureSelfGamePoolDay(ctx, tx, event.AppCode, scope.tz, scope.day, gameID, direction, reason, nowMS); err != nil {
return err
}
if _, 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_tz = ? AND stat_day = ? AND game_id = ? AND direction = ? AND reason = ?
`, inCoin, outCoin, event.BalanceAfter, nowMS, appcode.Normalize(event.AppCode), scope.tz, scope.day, gameID, direction, reason); err != nil {
return err
}
}
return nil
})
}
func consumeSelfGameSettledParticipant(ctx context.Context, tx *sql.Tx, event SelfGameMatchEvent, participant SelfGameMatchParticipantEvent, statTZ string, 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, statTZ, 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_tz = ? 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), normalizeStatTZ(statTZ), 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, statTZ, 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_tz = ? 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), normalizeStatTZ(statTZ), 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 isMissingIndexError(err error) bool {
var mysqlErr *mysqlerr.MySQLError
return errors.As(err, &mysqlErr) && mysqlErr.Number == 1091
}
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, statTZ string, day string, countryID int64, regionID int64, userID int64, occurredAtMS int64, nowMS int64) error {
if err := ensureAppDay(ctx, tx, app, statTZ, day, countryID, regionID, nowMS); err != nil {
return err
}
affected, err := insertUnique(ctx, tx, `
INSERT IGNORE INTO stat_user_day_activity (app_code, stat_tz, stat_day, country_id, region_id, user_id, first_active_at_ms)
VALUES (?, ?, ?, ?, ?, ?, ?)
`, appcode.Normalize(app), normalizeStatTZ(statTZ), 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_tz = ? AND stat_day = ? AND country_id = ? AND region_id = ?
`, nowMS, appcode.Normalize(app), normalizeStatTZ(statTZ), day, countryID, regionID)
return err
}
func upsertUserDimension(ctx context.Context, tx *sql.Tx, app string, userID int64, countryID int64, regionID int64, nowMS int64) error {
if userID <= 0 {
return nil
}
_, err := tx.ExecContext(ctx, `
INSERT INTO stat_user_dimension (app_code, user_id, country_id, region_id, updated_at_ms)
VALUES (?, ?, ?, ?, ?)
ON DUPLICATE KEY UPDATE
country_id = CASE WHEN VALUES(country_id) > 0 THEN VALUES(country_id) ELSE country_id END,
region_id = CASE WHEN VALUES(region_id) > 0 THEN VALUES(region_id) ELSE region_id END,
updated_at_ms = VALUES(updated_at_ms)
`, appcode.Normalize(app), userID, normalizeID(countryID), normalizeID(regionID), nowMS)
return err
}
func (r *Repository) dimensionForUser(ctx context.Context, tx *sql.Tx, app string, userID int64, countryID int64, regionID int64) (int64, int64, error) {
countryID, regionID = normalizeDimension(countryID, regionID)
if (countryID > 0 && regionID > 0) || userID <= 0 {
return countryID, regionID, nil
}
var storedCountryID, storedRegionID int64
err := tx.QueryRowContext(ctx, `
SELECT country_id, region_id
FROM stat_user_dimension
WHERE app_code = ? AND user_id = ?
`, appcode.Normalize(app), userID).Scan(&storedCountryID, &storedRegionID)
if err != nil && err != sql.ErrNoRows {
return 0, 0, err
}
if countryID == 0 {
countryID = storedCountryID
}
if regionID == 0 {
regionID = storedRegionID
}
countryID, regionID = normalizeDimension(countryID, regionID)
return countryID, regionID, nil
}
func isConsumedCoinBizType(value string) bool {
switch strings.ToLower(strings.TrimSpace(value)) {
case "direct_gift_debit", "red_packet_create", "wheel_draw_debit":
return true
default:
return false
}
}
func isOutputCoinBizType(value string) bool {
switch strings.ToLower(strings.TrimSpace(value)) {
case "red_packet_claim", "red_packet_refund":
return true
default:
return false
}
}
func isHostAgencySalaryAssetType(value string) bool {
switch strings.ToUpper(strings.TrimSpace(value)) {
case "HOST_SALARY_USD", "AGENCY_SALARY_USD":
return true
default:
return false
}
}
func isOutputCoinRewardEventType(value string) bool {
switch strings.TrimSpace(value) {
case "WalletWheelRewardCredited":
return true
default:
return false
}
}
func isPlatformGrantCommandID(value string) bool {
normalized := strings.ToLower(strings.TrimSpace(value))
if normalized == "" {
return false
}
switch {
case strings.HasPrefix(normalized, "weekly_star:"):
return true
case strings.HasPrefix(normalized, "cp_weekly_rank:"):
return true
case strings.HasPrefix(normalized, "room_turnover_reward:"):
return true
case strings.HasPrefix(normalized, "wcheckin_"):
return true
case strings.HasPrefix(normalized, "cmd_room_rocket_grant_"):
return true
case strings.HasPrefix(normalized, "wtask_"), strings.HasPrefix(normalized, "wfr_"), strings.HasPrefix(normalized, "wcr_"), strings.HasPrefix(normalized, "wreg_"):
return true
default:
return false
}
}
func statDateScopesFromContext(ctx context.Context, statDate string) []statDayScope {
day := strings.TrimSpace(statDate)
if ctx != nil {
if statTZ, ok := ctx.Value(statTZScopeContextKey{}).(string); ok {
return []statDayScope{{tz: normalizeStatTZ(statTZ), day: day}}
}
}
return []statDayScope{{tz: StatTZUTC, day: day}, {tz: StatTZAsiaShanghai, day: day}}
}
func micOnlineBefore(ctx context.Context, tx *sql.Tx, app string, statTZ string, day string, userID int64) (int64, error) {
var before int64
err := tx.QueryRowContext(ctx, `
SELECT mic_online_ms
FROM stat_user_day_mic
WHERE app_code = ? AND stat_tz = ? AND stat_day = ? AND user_id = ?
FOR UPDATE
`, appcode.Normalize(app), normalizeStatTZ(statTZ), day, userID).Scan(&before)
if err == sql.ErrNoRows {
return 0, nil
}
return before, err
}
func ensureAppDay(ctx context.Context, tx *sql.Tx, app string, statTZ string, day string, countryID int64, regionID int64, nowMS int64) error {
_, err := tx.ExecContext(ctx, `
INSERT INTO stat_app_day_country (app_code, stat_tz, stat_day, country_id, region_id, updated_at_ms)
VALUES (?, ?, ?, ?, ?, ?)
ON DUPLICATE KEY UPDATE
updated_at_ms = VALUES(updated_at_ms)
`, appcode.Normalize(app), normalizeStatTZ(statTZ), day, countryID, regionID, nowMS)
return err
}
func ensureGameDay(ctx context.Context, tx *sql.Tx, app string, statTZ 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_tz, 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), normalizeStatTZ(statTZ), day, countryID, regionID, platformCode, gameID, nowMS)
return err
}
func ensureLuckyGiftPoolDay(ctx context.Context, tx *sql.Tx, app string, statTZ 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_tz, 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), normalizeStatTZ(statTZ), day, countryID, regionID, strings.TrimSpace(poolID), nowMS)
return err
}
func insertPlatformGrantCoinEvent(ctx context.Context, tx *sql.Tx, app string, statTZ string, day string, eventID string, eventType string, userID int64, countryID int64, regionID int64, amount int64, occurredAtMS int64, nowMS int64) error {
eventID = strings.TrimSpace(eventID)
eventType = strings.TrimSpace(eventType)
amount = normalizeID(amount)
if eventID == "" || userID <= 0 || amount <= 0 {
return nil
}
if eventType == "" {
eventType = "WalletPlatformGrantCoinCredited"
}
// withEvent 已经保证同一个钱包事件只聚合一次；这里仍用 INSERT IGNORE，让补偿消费或历史修复遇到旧明细时不回滚当天汇总事务。
_, err := tx.ExecContext(ctx, `
INSERT IGNORE INTO stat_platform_grant_coin_events (
app_code, stat_tz, stat_day, event_id, event_type, user_id, country_id, region_id,
amount, occurred_at_ms, updated_at_ms
) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
`, appcode.Normalize(app), normalizeStatTZ(statTZ), day, eventID, eventType, userID, countryID, regionID, amount, occurredAtMS, nowMS)
return err
}
func ensureSelfGameH5EventDay(ctx context.Context, tx *sql.Tx, app string, statTZ 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_tz, 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), normalizeStatTZ(statTZ), day, countryID, regionID, gameID, eventName, language, clientVersion, h5Version, entrySource, nowMS)
return err
}
func ensureSelfGameMatchDay(ctx context.Context, tx *sql.Tx, app string, statTZ 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_tz, 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), normalizeStatTZ(statTZ), day, countryID, regionID, gameID, stakeCoin, nowMS)
return err
}
func ensureSelfGamePoolDay(ctx context.Context, tx *sql.Tx, app string, statTZ 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_tz, stat_day, game_id, direction, reason, updated_at_ms)
VALUES (?, ?, ?, ?, ?, ?, ?)
ON DUPLICATE KEY UPDATE
updated_at_ms = VALUES(updated_at_ms)
`, appcode.Normalize(app), normalizeStatTZ(statTZ), day, gameID, direction, reason, nowMS)
return err
}
func ensureSelfGameParticipantDay(ctx context.Context, tx *sql.Tx, app string, statTZ 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_tz, 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), normalizeStatTZ(statTZ), day, countryID, regionID, gameID, stakeCoin, participantType, result, gesture, dicePoint, nowMS)
return err
}
func ensureSelfGameUserDay(ctx context.Context, tx *sql.Tx, app string, statTZ 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_tz, 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), normalizeStatTZ(statTZ), 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 statDayIn(ms, StatTZUTC)
}
func normalizeID(id int64) int64 {
if id < 0 {
return 0
}
return id
}
func normalizeShortText(value string) string {
return strings.TrimSpace(value)
}
func normalizeAppTrackingEvent(ctx context.Context, event AppTrackingEvent, receivedAtMS int64) (AppTrackingEvent, string, error) {
var err error
if strings.TrimSpace(event.AppCode) == "" {
event.AppCode = appcode.FromContext(ctx)
}
event.AppCode = appcode.Normalize(event.AppCode)
event.EventID, err = normalizeAppTrackingText(event.EventID, 160)
if err != nil {
return AppTrackingEvent{}, "", err
}
event.EventName, err = normalizeAppTrackingText(event.EventName, 96)
if err != nil {
return AppTrackingEvent{}, "", err
}
if event.EventID == "" || event.EventName == "" {
return AppTrackingEvent{}, "", xerr.New(xerr.InvalidArgument, "app tracking event_id and event_name are required")
}
event.EventType, err = normalizeAppTrackingText(event.EventType, 64)
if err != nil {
return AppTrackingEvent{}, "", err
}
event.Screen, err = normalizeAppTrackingText(event.Screen, 128)
if err != nil {
return AppTrackingEvent{}, "", err
}
event.TargetType, err = normalizeAppTrackingText(event.TargetType, 64)
if err != nil {
return AppTrackingEvent{}, "", err
}
event.TargetID, err = normalizeAppTrackingText(event.TargetID, 128)
if err != nil {
return AppTrackingEvent{}, "", err
}
event.DeviceID, err = normalizeAppTrackingText(event.DeviceID, 128)
if err != nil {
return AppTrackingEvent{}, "", err
}
if event.UserID <= 0 && event.DeviceID == "" {
return AppTrackingEvent{}, "", xerr.New(xerr.InvalidArgument, "anonymous app tracking event requires device_id")
}
event.SessionID, err = normalizeAppTrackingText(event.SessionID, 160)
if err != nil {
return AppTrackingEvent{}, "", err
}
event.Platform, err = normalizeAppTrackingText(strings.ToLower(event.Platform), 32)
if err != nil {
return AppTrackingEvent{}, "", err
}
event.AppVersion, err = normalizeAppTrackingText(event.AppVersion, 64)
if err != nil {
return AppTrackingEvent{}, "", err
}
event.Language, err = normalizeAppTrackingText(event.Language, 32)
if err != nil {
return AppTrackingEvent{}, "", err
}
event.Timezone, err = normalizeAppTrackingText(event.Timezone, 64)
if err != nil {
return AppTrackingEvent{}, "", err
}
event.ErrorCode, err = normalizeAppTrackingText(event.ErrorCode, 64)
if err != nil {
return AppTrackingEvent{}, "", err
}
event.UserID = normalizeID(event.UserID)
event.CountryID, event.RegionID = normalizeDimension(event.CountryID, event.RegionID)
event.DurationMS = normalizeID(event.DurationMS)
if event.OccurredAtMS <= 0 {
event.OccurredAtMS = receivedAtMS
}
propertiesJSON, err := normalizeAppTrackingProperties(event.Properties)
if err != nil {
return AppTrackingEvent{}, "", err
}
return event, propertiesJSON, nil
}
func normalizeAppTrackingText(value string, maxBytes int) (string, error) {
value = strings.TrimSpace(value)
if maxBytes > 0 && len(value) > maxBytes {
return "", xerr.New(xerr.InvalidArgument, "app tracking text field is too long")
}
return value, nil
}
func normalizeAppTrackingProperties(raw json.RawMessage) (string, error) {
trimmed := bytes.TrimSpace(raw)
if len(trimmed) == 0 || bytes.Equal(trimmed, []byte("null")) {
return "{}", nil
}
if len(trimmed) > appTrackingMaxPropertiesBytes || !json.Valid(trimmed) {
return "", xerr.New(xerr.InvalidArgument, "app tracking properties_json is invalid")
}
return string(trimmed), nil
}
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 ""
}
Reference in New Issue View Git Blame Copy Permalink