656 lines
23 KiB
Go
656 lines
23 KiB
Go
package main
|
||
|
||
import (
|
||
"context"
|
||
"database/sql"
|
||
"flag"
|
||
"fmt"
|
||
"log"
|
||
"os"
|
||
"sort"
|
||
"strings"
|
||
"time"
|
||
|
||
_ "github.com/go-sql-driver/mysql"
|
||
"hyapp/pkg/appcode"
|
||
)
|
||
|
||
type countryRegion struct {
|
||
countryID int64
|
||
regionID int64
|
||
}
|
||
|
||
type appDayKey struct {
|
||
day string
|
||
countryID int64
|
||
regionID int64
|
||
}
|
||
|
||
type appDayAgg struct {
|
||
newUsers int64
|
||
activeUsers int64
|
||
paidUsers int64
|
||
rechargeUSDMinor int64
|
||
newUserRechargeUSDMinor int64
|
||
coinSellerRechargeUSDMinor int64
|
||
googleRechargeUSDMinor int64
|
||
coinSellerTransferCoin int64
|
||
}
|
||
|
||
type userRegistrationRow struct {
|
||
userID int64
|
||
registeredDay string
|
||
dim countryRegion
|
||
createdAtMS int64
|
||
}
|
||
|
||
type payerRow struct {
|
||
day string
|
||
userID int64
|
||
dim countryRegion
|
||
firstMS int64
|
||
}
|
||
|
||
type activeRow struct {
|
||
day string
|
||
userID int64
|
||
dim countryRegion
|
||
firstMS int64
|
||
}
|
||
|
||
type rebuildState struct {
|
||
appDays map[appDayKey]*appDayAgg
|
||
registrations []userRegistrationRow
|
||
registeredByUserDay map[string]struct{}
|
||
payersByKey map[string]payerRow
|
||
activeByKey map[string]activeRow
|
||
userIDs map[int64]struct{}
|
||
userDims map[int64]countryRegion
|
||
}
|
||
|
||
func main() {
|
||
ctx := context.Background()
|
||
var (
|
||
appFlag = flag.String("app", envDefault("APP_CODE", "lalu"), "app_code to backfill")
|
||
tzFlag = flag.String("timezone", envDefault("BACKFILL_TIMEZONE", "UTC"), "day boundary timezone, e.g. UTC or Asia/Shanghai")
|
||
startFlag = flag.String("start-day", os.Getenv("START_DAY"), "inclusive yyyy-mm-dd in timezone; defaults to last 7 days")
|
||
endFlag = flag.String("end-day", os.Getenv("END_DAY"), "inclusive yyyy-mm-dd in timezone; defaults to today")
|
||
statsDSNFlag = flag.String("statistics-dsn", firstNonEmpty(os.Getenv("STATISTICS_MYSQL_DSN"), os.Getenv("MYSQL_DSN")), "statistics mysql dsn")
|
||
userDSNFlag = flag.String("user-dsn", os.Getenv("USER_MYSQL_DSN"), "user-service mysql dsn")
|
||
walletDSNFlag = flag.String("wallet-dsn", os.Getenv("WALLET_MYSQL_DSN"), "wallet-service mysql dsn")
|
||
applyFlag = flag.Bool("apply", envBool("APPLY"), "apply replacement; default dry-run")
|
||
)
|
||
flag.Parse()
|
||
|
||
if strings.TrimSpace(*statsDSNFlag) == "" || strings.TrimSpace(*userDSNFlag) == "" || strings.TrimSpace(*walletDSNFlag) == "" {
|
||
log.Fatal("statistics-dsn, user-dsn and wallet-dsn are required")
|
||
}
|
||
loc, err := time.LoadLocation(strings.TrimSpace(*tzFlag))
|
||
if err != nil {
|
||
log.Fatalf("load timezone: %v", err)
|
||
}
|
||
startDay, endDay, windowStartMS, windowEndMS, err := resolveWindow(*startFlag, *endFlag, loc)
|
||
if err != nil {
|
||
log.Fatal(err)
|
||
}
|
||
app := appcode.Normalize(*appFlag)
|
||
|
||
statsDB := mustOpen(ctx, *statsDSNFlag)
|
||
defer statsDB.Close()
|
||
userDB := mustOpen(ctx, *userDSNFlag)
|
||
defer userDB.Close()
|
||
walletDB := mustOpen(ctx, *walletDSNFlag)
|
||
defer walletDB.Close()
|
||
|
||
state := &rebuildState{
|
||
appDays: map[appDayKey]*appDayAgg{},
|
||
registeredByUserDay: map[string]struct{}{},
|
||
payersByKey: map[string]payerRow{},
|
||
activeByKey: map[string]activeRow{},
|
||
userIDs: map[int64]struct{}{},
|
||
userDims: map[int64]countryRegion{},
|
||
}
|
||
// 第一阶段只收集窗口内会参与重算的 user_id，后续统一批量加载用户国家和区域，避免按充值/活跃记录逐条访问 user-service 数据库。
|
||
if err := collectRegistrationUsers(ctx, userDB, state, app, windowStartMS, windowEndMS); err != nil {
|
||
log.Fatalf("collect registrations: %v", err)
|
||
}
|
||
if err := collectActiveUserIDs(ctx, statsDB, state, app, startDay, endDay); err != nil {
|
||
log.Fatalf("collect active ids: %v", err)
|
||
}
|
||
if err := collectRechargeUserIDs(ctx, walletDB, state, app, windowStartMS, windowEndMS); err != nil {
|
||
log.Fatalf("collect recharge ids: %v", err)
|
||
}
|
||
if err := loadUserDims(ctx, userDB, state, app); err != nil {
|
||
log.Fatalf("load user dims: %v", err)
|
||
}
|
||
if err := rebuildRegistrations(ctx, userDB, state, app, windowStartMS, windowEndMS, loc); err != nil {
|
||
log.Fatalf("rebuild registrations: %v", err)
|
||
}
|
||
if err := rebuildActive(ctx, statsDB, state, app, startDay, endDay); err != nil {
|
||
log.Fatalf("rebuild active: %v", err)
|
||
}
|
||
if err := rebuildWalletRecharge(ctx, walletDB, state, app, windowStartMS, windowEndMS, loc); err != nil {
|
||
log.Fatalf("rebuild wallet recharge: %v", err)
|
||
}
|
||
if err := rebuildCoinSellerStock(ctx, walletDB, state, app, windowStartMS, windowEndMS, loc); err != nil {
|
||
log.Fatalf("rebuild coin seller stock: %v", err)
|
||
}
|
||
|
||
if err := stageAndMaybeApply(ctx, statsDB, state, app, startDay, endDay, *applyFlag); err != nil {
|
||
log.Fatalf("stage/apply: %v", err)
|
||
}
|
||
mode := "dry-run"
|
||
if *applyFlag {
|
||
mode = "applied"
|
||
}
|
||
fmt.Printf("%s app=%s timezone=%s window=%s..%s app_day_rows=%d registrations=%d active=%d payers=%d\n",
|
||
mode, app, loc.String(), startDay, endDay, len(state.appDays), len(state.registrations), len(state.activeByKey), len(state.payersByKey))
|
||
}
|
||
|
||
func envDefault(key string, fallback string) string {
|
||
if value := strings.TrimSpace(os.Getenv(key)); value != "" {
|
||
return value
|
||
}
|
||
return fallback
|
||
}
|
||
|
||
func envBool(key string) bool {
|
||
switch strings.ToLower(strings.TrimSpace(os.Getenv(key))) {
|
||
case "1", "true", "yes", "y":
|
||
return true
|
||
default:
|
||
return false
|
||
}
|
||
}
|
||
|
||
func firstNonEmpty(values ...string) string {
|
||
for _, value := range values {
|
||
if strings.TrimSpace(value) != "" {
|
||
return strings.TrimSpace(value)
|
||
}
|
||
}
|
||
return ""
|
||
}
|
||
|
||
func mustOpen(ctx context.Context, dsn string) *sql.DB {
|
||
db, err := sql.Open("mysql", strings.TrimSpace(dsn))
|
||
if err != nil {
|
||
log.Fatal(err)
|
||
}
|
||
if err := db.PingContext(ctx); err != nil {
|
||
log.Fatal(err)
|
||
}
|
||
return db
|
||
}
|
||
|
||
func resolveWindow(rawStart, rawEnd string, loc *time.Location) (string, string, int64, int64, error) {
|
||
now := time.Now().In(loc)
|
||
endDayTime := time.Date(now.Year(), now.Month(), now.Day(), 0, 0, 0, 0, loc)
|
||
startDayTime := endDayTime.AddDate(0, 0, -6)
|
||
var err error
|
||
if strings.TrimSpace(rawEnd) != "" {
|
||
endDayTime, err = parseDay(rawEnd, loc)
|
||
if err != nil {
|
||
return "", "", 0, 0, err
|
||
}
|
||
}
|
||
if strings.TrimSpace(rawStart) != "" {
|
||
startDayTime, err = parseDay(rawStart, loc)
|
||
if err != nil {
|
||
return "", "", 0, 0, err
|
||
}
|
||
}
|
||
if startDayTime.After(endDayTime) {
|
||
return "", "", 0, 0, fmt.Errorf("start-day must not be after end-day")
|
||
}
|
||
windowEnd := endDayTime.AddDate(0, 0, 1)
|
||
return formatDay(startDayTime), formatDay(endDayTime), startDayTime.UTC().UnixMilli(), windowEnd.UTC().UnixMilli(), nil
|
||
}
|
||
|
||
func parseDay(value string, loc *time.Location) (time.Time, error) {
|
||
parsed, err := time.ParseInLocation("2006-01-02", strings.TrimSpace(value), loc)
|
||
if err != nil {
|
||
return time.Time{}, err
|
||
}
|
||
return time.Date(parsed.Year(), parsed.Month(), parsed.Day(), 0, 0, 0, 0, loc), nil
|
||
}
|
||
|
||
func dayFromMS(ms int64, loc *time.Location) string {
|
||
return formatDay(time.UnixMilli(ms).In(loc))
|
||
}
|
||
|
||
func formatDay(value time.Time) string {
|
||
return value.Format("2006-01-02")
|
||
}
|
||
|
||
func collectRegistrationUsers(ctx context.Context, db *sql.DB, state *rebuildState, app string, startMS, endMS int64) error {
|
||
// 全站机器人使用 users 资料参与游戏展示，但不是自然注册用户；补数入口必须和实时 UserRegistered 口径一致排除。
|
||
rows, err := db.QueryContext(ctx, `
|
||
SELECT u.user_id
|
||
FROM users u
|
||
WHERE u.app_code = ?
|
||
AND u.created_at_ms >= ?
|
||
AND u.created_at_ms < ?
|
||
AND LOWER(COALESCE(u.source, '')) <> 'game_robot'`,
|
||
app, startMS, endMS)
|
||
if err != nil {
|
||
return err
|
||
}
|
||
defer rows.Close()
|
||
for rows.Next() {
|
||
var userID int64
|
||
if err := rows.Scan(&userID); err != nil {
|
||
return err
|
||
}
|
||
state.userIDs[userID] = struct{}{}
|
||
}
|
||
return rows.Err()
|
||
}
|
||
|
||
func collectActiveUserIDs(ctx context.Context, db *sql.DB, state *rebuildState, app, startDay, endDay string) error {
|
||
rows, err := db.QueryContext(ctx, `
|
||
SELECT DISTINCT user_id
|
||
FROM stat_user_day_activity
|
||
WHERE app_code = ? AND stat_day BETWEEN ? AND ?`,
|
||
app, startDay, endDay)
|
||
if err != nil {
|
||
return err
|
||
}
|
||
defer rows.Close()
|
||
for rows.Next() {
|
||
var userID int64
|
||
if err := rows.Scan(&userID); err != nil {
|
||
return err
|
||
}
|
||
state.userIDs[userID] = struct{}{}
|
||
}
|
||
return rows.Err()
|
||
}
|
||
|
||
func collectRechargeUserIDs(ctx context.Context, db *sql.DB, state *rebuildState, app string, startMS, endMS int64) error {
|
||
for _, query := range []string{
|
||
`SELECT DISTINCT user_id FROM wallet_recharge_records WHERE app_code = ? AND created_at_ms >= ? AND created_at_ms < ?`,
|
||
`SELECT DISTINCT seller_user_id FROM coin_seller_stock_records WHERE app_code = ? AND counts_as_seller_recharge = 1 AND created_at_ms >= ? AND created_at_ms < ?`,
|
||
} {
|
||
rows, err := db.QueryContext(ctx, query, app, startMS, endMS)
|
||
if err != nil {
|
||
return err
|
||
}
|
||
for rows.Next() {
|
||
var userID int64
|
||
if err := rows.Scan(&userID); err != nil {
|
||
rows.Close()
|
||
return err
|
||
}
|
||
state.userIDs[userID] = struct{}{}
|
||
}
|
||
if err := rows.Close(); err != nil {
|
||
return err
|
||
}
|
||
}
|
||
return nil
|
||
}
|
||
|
||
func loadUserDims(ctx context.Context, db *sql.DB, state *rebuildState, app string) error {
|
||
userIDs := make([]int64, 0, len(state.userIDs))
|
||
for userID := range state.userIDs {
|
||
userIDs = append(userIDs, userID)
|
||
}
|
||
sort.Slice(userIDs, func(i, j int) bool { return userIDs[i] < userIDs[j] })
|
||
for start := 0; start < len(userIDs); start += 500 {
|
||
end := start + 500
|
||
if end > len(userIDs) {
|
||
end = len(userIDs)
|
||
}
|
||
placeholders := strings.TrimRight(strings.Repeat("?,", end-start), ",")
|
||
args := []any{app}
|
||
for _, userID := range userIDs[start:end] {
|
||
args = append(args, userID)
|
||
}
|
||
rows, err := db.QueryContext(ctx, `
|
||
SELECT u.user_id, COALESCE(c.country_id, 0), COALESCE(u.region_id, 0)
|
||
FROM users u
|
||
LEFT JOIN countries c ON c.app_code = u.app_code AND c.country_code = u.country
|
||
WHERE u.app_code = ? AND u.user_id IN (`+placeholders+`)`, args...)
|
||
if err != nil {
|
||
return err
|
||
}
|
||
for rows.Next() {
|
||
var userID int64
|
||
var dim countryRegion
|
||
if err := rows.Scan(&userID, &dim.countryID, &dim.regionID); err != nil {
|
||
rows.Close()
|
||
return err
|
||
}
|
||
state.userDims[userID] = dim
|
||
}
|
||
if err := rows.Close(); err != nil {
|
||
return err
|
||
}
|
||
}
|
||
return nil
|
||
}
|
||
|
||
func rebuildRegistrations(ctx context.Context, db *sql.DB, state *rebuildState, app string, startMS, endMS int64, loc *time.Location) error {
|
||
// 注册国家以 users.country 映射出的真实 country_id 为准；region_id 只作为独立维度写入，不能再兜底成国家。
|
||
// source=game_robot 是后台全站机器人池账号，不能进入新增用户、注册 cohort 或留存口径。
|
||
rows, err := db.QueryContext(ctx, `
|
||
SELECT u.user_id, COALESCE(c.country_id, 0), COALESCE(u.region_id, 0), u.created_at_ms
|
||
FROM users u
|
||
LEFT JOIN countries c ON c.app_code = u.app_code AND c.country_code = u.country
|
||
WHERE u.app_code = ?
|
||
AND u.created_at_ms >= ?
|
||
AND u.created_at_ms < ?
|
||
AND LOWER(COALESCE(u.source, '')) <> 'game_robot'`,
|
||
app, startMS, endMS)
|
||
if err != nil {
|
||
return err
|
||
}
|
||
defer rows.Close()
|
||
for rows.Next() {
|
||
var row userRegistrationRow
|
||
if err := rows.Scan(&row.userID, &row.dim.countryID, &row.dim.regionID, &row.createdAtMS); err != nil {
|
||
return err
|
||
}
|
||
row.registeredDay = dayFromMS(row.createdAtMS, loc)
|
||
state.registrations = append(state.registrations, row)
|
||
state.registeredByUserDay[userDayKey(row.registeredDay, row.userID)] = struct{}{}
|
||
agg := state.ensureAgg(row.registeredDay, row.dim)
|
||
agg.newUsers++
|
||
}
|
||
return rows.Err()
|
||
}
|
||
|
||
func rebuildActive(ctx context.Context, db *sql.DB, state *rebuildState, app, startDay, endDay string) error {
|
||
// 活跃的原始房间事件已经被实时消费进 stat_user_day_activity；离线回填只重刷该表的国家/区域快照，避免在大窗口内扫描房间流水。
|
||
rows, err := db.QueryContext(ctx, `
|
||
SELECT DATE_FORMAT(stat_day, '%Y-%m-%d'), user_id, MIN(first_active_at_ms)
|
||
FROM stat_user_day_activity
|
||
WHERE app_code = ? AND stat_day BETWEEN ? AND ?
|
||
GROUP BY stat_day, user_id`,
|
||
app, startDay, endDay)
|
||
if err != nil {
|
||
return err
|
||
}
|
||
defer rows.Close()
|
||
for rows.Next() {
|
||
var day string
|
||
var userID, firstMS int64
|
||
if err := rows.Scan(&day, &userID, &firstMS); err != nil {
|
||
return err
|
||
}
|
||
dim := state.userDims[userID]
|
||
key := fmt.Sprintf("%s:%d", day, userID)
|
||
if _, exists := state.activeByKey[key]; exists {
|
||
continue
|
||
}
|
||
state.activeByKey[key] = activeRow{day: day, userID: userID, dim: dim, firstMS: firstMS}
|
||
state.ensureAgg(day, dim).activeUsers++
|
||
}
|
||
return rows.Err()
|
||
}
|
||
|
||
func rebuildWalletRecharge(ctx context.Context, db *sql.DB, state *rebuildState, app string, startMS, endMS int64, loc *time.Location) error {
|
||
// wallet_recharge_records 是到账事实表；Google 也会写入该表，payment_orders 只是 provider 审计，不能在回填里再扫一次导致双算。
|
||
// 币商向用户转账只代表金币充值用户，不产生用户 USDT 充值；Google Play 则按记录里的 usd_minor_amount 计入用户充值。
|
||
rows, err := db.QueryContext(ctx, `
|
||
SELECT rr.user_id, rr.recharge_sequence, rr.coin_amount, rr.usd_minor_amount, rr.created_at_ms, wt.biz_type
|
||
FROM wallet_recharge_records rr
|
||
JOIN wallet_transactions wt ON wt.app_code = rr.app_code AND wt.transaction_id = rr.transaction_id
|
||
WHERE rr.app_code = ? AND rr.created_at_ms >= ? AND rr.created_at_ms < ?`,
|
||
app, startMS, endMS)
|
||
if err != nil {
|
||
return err
|
||
}
|
||
defer rows.Close()
|
||
for rows.Next() {
|
||
var userID, sequence, coinAmount, usdMinor, createdAtMS int64
|
||
var bizType string
|
||
if err := rows.Scan(&userID, &sequence, &coinAmount, &usdMinor, &createdAtMS, &bizType); err != nil {
|
||
return err
|
||
}
|
||
day := dayFromMS(createdAtMS, loc)
|
||
dim := state.userDims[userID]
|
||
state.addPayer(day, userID, dim, createdAtMS)
|
||
agg := state.ensureAgg(day, dim)
|
||
switch strings.ToLower(strings.TrimSpace(bizType)) {
|
||
case "coin_seller_transfer":
|
||
agg.coinSellerTransferCoin += coinAmount
|
||
case "google_play_recharge":
|
||
agg.rechargeUSDMinor += usdMinor
|
||
agg.googleRechargeUSDMinor += usdMinor
|
||
default:
|
||
if usdMinor > 0 {
|
||
agg.rechargeUSDMinor += usdMinor
|
||
}
|
||
}
|
||
if sequence == 1 && usdMinor > 0 {
|
||
agg.newUserRechargeUSDMinor += usdMinor
|
||
}
|
||
}
|
||
return rows.Err()
|
||
}
|
||
|
||
func rebuildCoinSellerStock(ctx context.Context, db *sql.DB, state *rebuildState, app string, startMS, endMS int64, loc *time.Location) error {
|
||
// 币商进货是平台收到的 USDT 充值，按 seller 的国家/区域进入总充值，但不写入用户金币转账金额。
|
||
rows, err := db.QueryContext(ctx, `
|
||
SELECT seller_user_id, paid_amount_micro, created_at_ms
|
||
FROM coin_seller_stock_records
|
||
WHERE app_code = ? AND counts_as_seller_recharge = 1 AND paid_amount_micro > 0 AND created_at_ms >= ? AND created_at_ms < ?`,
|
||
app, startMS, endMS)
|
||
if err != nil {
|
||
return err
|
||
}
|
||
defer rows.Close()
|
||
for rows.Next() {
|
||
var sellerUserID, amountMicro, createdAtMS int64
|
||
if err := rows.Scan(&sellerUserID, &amountMicro, &createdAtMS); err != nil {
|
||
return err
|
||
}
|
||
usdMinor := amountMicro / 10_000
|
||
day := dayFromMS(createdAtMS, loc)
|
||
dim := state.userDims[sellerUserID]
|
||
agg := state.ensureAgg(day, dim)
|
||
agg.rechargeUSDMinor += usdMinor
|
||
agg.coinSellerRechargeUSDMinor += usdMinor
|
||
}
|
||
return rows.Err()
|
||
}
|
||
|
||
func (s *rebuildState) ensureAgg(day string, dim countryRegion) *appDayAgg {
|
||
key := appDayKey{day: day, countryID: dim.countryID, regionID: dim.regionID}
|
||
if s.appDays[key] == nil {
|
||
s.appDays[key] = &appDayAgg{}
|
||
}
|
||
return s.appDays[key]
|
||
}
|
||
|
||
func (s *rebuildState) addPayer(day string, userID int64, dim countryRegion, firstMS int64) {
|
||
key := fmt.Sprintf("%s:%d", day, userID)
|
||
if existing, exists := s.payersByKey[key]; exists && existing.firstMS <= firstMS {
|
||
return
|
||
}
|
||
if _, exists := s.payersByKey[key]; !exists {
|
||
s.ensureAgg(day, dim).paidUsers++
|
||
}
|
||
s.payersByKey[key] = payerRow{day: day, userID: userID, dim: dim, firstMS: firstMS}
|
||
}
|
||
|
||
func (s *rebuildState) userRegisteredOn(userID int64, day string) bool {
|
||
_, ok := s.registeredByUserDay[userDayKey(day, userID)]
|
||
return ok
|
||
}
|
||
|
||
func userDayKey(day string, userID int64) string {
|
||
return fmt.Sprintf("%s:%d", day, userID)
|
||
}
|
||
|
||
func stageAndMaybeApply(ctx context.Context, db *sql.DB, state *rebuildState, app, startDay, endDay string, apply bool) error {
|
||
conn, err := db.Conn(ctx)
|
||
if err != nil {
|
||
return err
|
||
}
|
||
defer conn.Close()
|
||
tx, err := conn.BeginTx(ctx, nil)
|
||
if err != nil {
|
||
return err
|
||
}
|
||
defer tx.Rollback()
|
||
if err := createTempTables(ctx, tx); err != nil {
|
||
return err
|
||
}
|
||
if err := insertStageRows(ctx, tx, state, app); err != nil {
|
||
return err
|
||
}
|
||
if err := validateStage(ctx, tx); err != nil {
|
||
return err
|
||
}
|
||
if !apply {
|
||
return tx.Rollback()
|
||
}
|
||
nowMS := time.Now().UTC().UnixMilli()
|
||
// 本 job 只覆盖能从当前源表确定的近 7 天列；礼物、游戏、MifaPay 等不在本次源表范围内的指标保留原值，避免修国家维度时误删其他业务数据。
|
||
if _, err := tx.ExecContext(ctx, `
|
||
UPDATE stat_app_day_country
|
||
SET new_users = 0, active_users = 0, paid_users = 0,
|
||
recharge_usd_minor = 0, new_user_recharge_usd_minor = 0,
|
||
coin_seller_recharge_usd_minor = 0,
|
||
google_recharge_usd_minor = 0, coin_seller_transfer_coin = 0,
|
||
updated_at_ms = ?
|
||
WHERE app_code = ? AND stat_day BETWEEN ? AND ?`,
|
||
nowMS, app, startDay, endDay); err != nil {
|
||
return err
|
||
}
|
||
for _, statement := range []string{
|
||
`DELETE FROM stat_user_registration WHERE app_code = ? AND registered_day BETWEEN ? AND ?`,
|
||
`DELETE FROM stat_user_day_activity WHERE app_code = ? AND stat_day BETWEEN ? AND ?`,
|
||
`DELETE FROM stat_recharge_day_payers WHERE app_code = ? AND stat_day BETWEEN ? AND ?`,
|
||
} {
|
||
if _, err := tx.ExecContext(ctx, statement, app, startDay, endDay); err != nil {
|
||
return err
|
||
}
|
||
}
|
||
if _, err := tx.ExecContext(ctx, `
|
||
INSERT INTO stat_app_day_country (
|
||
app_code, stat_day, country_id, region_id, new_users, active_users, paid_users,
|
||
recharge_usd_minor, new_user_recharge_usd_minor, coin_seller_recharge_usd_minor,
|
||
google_recharge_usd_minor, coin_seller_transfer_coin, updated_at_ms
|
||
)
|
||
SELECT app_code, stat_day, country_id, region_id, new_users, active_users, paid_users,
|
||
recharge_usd_minor, new_user_recharge_usd_minor, coin_seller_recharge_usd_minor,
|
||
google_recharge_usd_minor, coin_seller_transfer_coin, ?
|
||
FROM tmp_stat_backfill_app_day
|
||
ON DUPLICATE KEY UPDATE
|
||
new_users = VALUES(new_users), active_users = VALUES(active_users), paid_users = VALUES(paid_users),
|
||
recharge_usd_minor = VALUES(recharge_usd_minor),
|
||
new_user_recharge_usd_minor = VALUES(new_user_recharge_usd_minor),
|
||
coin_seller_recharge_usd_minor = VALUES(coin_seller_recharge_usd_minor),
|
||
google_recharge_usd_minor = VALUES(google_recharge_usd_minor),
|
||
coin_seller_transfer_coin = VALUES(coin_seller_transfer_coin),
|
||
updated_at_ms = VALUES(updated_at_ms)`, nowMS); err != nil {
|
||
return err
|
||
}
|
||
if _, err := tx.ExecContext(ctx, `
|
||
INSERT INTO stat_user_registration (app_code, user_id, registered_day, country_id, region_id, registered_at_ms)
|
||
SELECT app_code, user_id, registered_day, country_id, region_id, registered_at_ms FROM tmp_stat_backfill_registration`); err != nil {
|
||
return err
|
||
}
|
||
if _, err := tx.ExecContext(ctx, `
|
||
INSERT INTO stat_user_day_activity (app_code, stat_day, country_id, region_id, user_id, first_active_at_ms)
|
||
SELECT app_code, stat_day, country_id, region_id, user_id, first_active_at_ms FROM tmp_stat_backfill_active`); err != nil {
|
||
return err
|
||
}
|
||
if _, err := tx.ExecContext(ctx, `
|
||
INSERT INTO stat_recharge_day_payers (app_code, stat_day, country_id, region_id, user_id, first_paid_at_ms)
|
||
SELECT app_code, stat_day, country_id, region_id, user_id, first_paid_at_ms FROM tmp_stat_backfill_payers`); err != nil {
|
||
return err
|
||
}
|
||
return tx.Commit()
|
||
}
|
||
|
||
func createTempTables(ctx context.Context, tx *sql.Tx) error {
|
||
statements := []string{
|
||
`CREATE TEMPORARY TABLE tmp_stat_backfill_app_day (
|
||
app_code VARCHAR(32) NOT NULL, stat_day DATE NOT NULL, country_id BIGINT NOT NULL, region_id BIGINT NOT NULL,
|
||
new_users BIGINT NOT NULL, active_users BIGINT NOT NULL, paid_users BIGINT NOT NULL,
|
||
recharge_usd_minor BIGINT NOT NULL, new_user_recharge_usd_minor BIGINT NOT NULL,
|
||
coin_seller_recharge_usd_minor BIGINT NOT NULL, google_recharge_usd_minor BIGINT NOT NULL,
|
||
coin_seller_transfer_coin BIGINT NOT NULL,
|
||
PRIMARY KEY(app_code, stat_day, country_id, region_id)
|
||
)`,
|
||
`CREATE TEMPORARY TABLE tmp_stat_backfill_registration (
|
||
app_code VARCHAR(32) NOT NULL, user_id BIGINT NOT NULL, registered_day DATE NOT NULL,
|
||
country_id BIGINT NOT NULL, region_id BIGINT NOT NULL, registered_at_ms BIGINT NOT NULL,
|
||
PRIMARY KEY(app_code, user_id)
|
||
)`,
|
||
`CREATE TEMPORARY TABLE tmp_stat_backfill_active (
|
||
app_code VARCHAR(32) NOT NULL, stat_day DATE NOT NULL, country_id BIGINT NOT NULL,
|
||
region_id BIGINT NOT NULL, user_id BIGINT NOT NULL, first_active_at_ms BIGINT NOT NULL,
|
||
PRIMARY KEY(app_code, stat_day, user_id)
|
||
)`,
|
||
`CREATE TEMPORARY TABLE tmp_stat_backfill_payers (
|
||
app_code VARCHAR(32) NOT NULL, stat_day DATE NOT NULL, country_id BIGINT NOT NULL,
|
||
region_id BIGINT NOT NULL, user_id BIGINT NOT NULL, first_paid_at_ms BIGINT NOT NULL,
|
||
PRIMARY KEY(app_code, stat_day, user_id)
|
||
)`,
|
||
}
|
||
for _, statement := range statements {
|
||
if _, err := tx.ExecContext(ctx, statement); err != nil {
|
||
return err
|
||
}
|
||
}
|
||
return nil
|
||
}
|
||
|
||
func insertStageRows(ctx context.Context, tx *sql.Tx, state *rebuildState, app string) error {
|
||
for key, agg := range state.appDays {
|
||
if _, err := tx.ExecContext(ctx, `
|
||
INSERT INTO tmp_stat_backfill_app_day (
|
||
app_code, stat_day, country_id, region_id, new_users, active_users, paid_users,
|
||
recharge_usd_minor, new_user_recharge_usd_minor, coin_seller_recharge_usd_minor,
|
||
google_recharge_usd_minor, coin_seller_transfer_coin
|
||
) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`,
|
||
app, key.day, key.countryID, key.regionID, agg.newUsers, agg.activeUsers, agg.paidUsers,
|
||
agg.rechargeUSDMinor, agg.newUserRechargeUSDMinor, agg.coinSellerRechargeUSDMinor,
|
||
agg.googleRechargeUSDMinor, agg.coinSellerTransferCoin); err != nil {
|
||
return err
|
||
}
|
||
}
|
||
for _, row := range state.registrations {
|
||
if _, err := tx.ExecContext(ctx, `
|
||
INSERT INTO tmp_stat_backfill_registration (app_code, user_id, registered_day, country_id, region_id, registered_at_ms)
|
||
VALUES (?, ?, ?, ?, ?, ?)`, app, row.userID, row.registeredDay, row.dim.countryID, row.dim.regionID, row.createdAtMS); err != nil {
|
||
return err
|
||
}
|
||
}
|
||
for _, row := range state.activeByKey {
|
||
if _, err := tx.ExecContext(ctx, `
|
||
INSERT INTO tmp_stat_backfill_active (app_code, stat_day, country_id, region_id, user_id, first_active_at_ms)
|
||
VALUES (?, ?, ?, ?, ?, ?)`, app, row.day, row.dim.countryID, row.dim.regionID, row.userID, row.firstMS); err != nil {
|
||
return err
|
||
}
|
||
}
|
||
for _, row := range state.payersByKey {
|
||
if _, err := tx.ExecContext(ctx, `
|
||
INSERT INTO tmp_stat_backfill_payers (app_code, stat_day, country_id, region_id, user_id, first_paid_at_ms)
|
||
VALUES (?, ?, ?, ?, ?, ?)`, app, row.day, row.dim.countryID, row.dim.regionID, row.userID, row.firstMS); err != nil {
|
||
return err
|
||
}
|
||
}
|
||
return nil
|
||
}
|
||
|
||
func validateStage(ctx context.Context, tx *sql.Tx) error {
|
||
var negativeCount int64
|
||
if err := tx.QueryRowContext(ctx, `
|
||
SELECT COUNT(*)
|
||
FROM tmp_stat_backfill_app_day
|
||
WHERE new_users < 0 OR active_users < 0 OR paid_users < 0 OR recharge_usd_minor < 0 OR coin_seller_transfer_coin < 0`).Scan(&negativeCount); err != nil {
|
||
return err
|
||
}
|
||
if negativeCount > 0 {
|
||
return fmt.Errorf("stage contains negative counters")
|
||
}
|
||
return nil
|
||
}
|