hyapp-server/tools/databi-real-flow/main.go

package main

import (
	"context"
	"database/sql"
	"encoding/json"
	"flag"
	"fmt"
	"log"
	"os"
	"strings"
	"time"

	gamev1 "hyapp.local/api/proto/game/v1"
	roomv1 "hyapp.local/api/proto/room/v1"
	userv1 "hyapp.local/api/proto/user/v1"
	walletv1 "hyapp.local/api/proto/wallet/v1"

	_ "github.com/go-sql-driver/mysql"
	"google.golang.org/grpc"
	"google.golang.org/grpc/credentials/insecure"
)

const (
	defaultAppCode = "lalu"
	countryCode    = "US"
	regionID       = int64(3)
	gameID         = "dice"
)

type config struct {
	appCode     string
	runID       string
	userDSN     string
	walletDSN   string
	userAddr    string
	walletAddr  string
	roomAddr    string
	gameAddr    string
	countryID   int64
	roomID      string
	admin       account
	sender      account
	host        account
	seller      account
	gamePlayer  account
	gameRival   account
	normalGift  string
	luckyGift   string
	superGift   string
	luckyPool   string
	superPool   string
	normalPrice int64
	luckyPrice  int64
	superPrice  int64
}

type account struct {
	userID        int64
	displayUserID string
	username      string
}

type runner struct {
	cfg          config
	userDB       *sql.DB
	walletDB     *sql.DB
	authClient   userv1.AuthServiceClient
	hostClient   userv1.UserHostAdminServiceClient
	walletClient walletv1.WalletServiceClient
	cronClient   walletv1.WalletCronServiceClient
	roomClient   roomv1.RoomCommandServiceClient
	gameAdmin    gamev1.GameAdminServiceClient
	gameApp      gamev1.GameAppServiceClient
}

func main() {
	var cfg config
	flag.StringVar(&cfg.appCode, "app", envDefault("APP_CODE", defaultAppCode), "app code")
	flag.StringVar(&cfg.runID, "run-id", envDefault("RUN_ID", ""), "stable run id")
	flag.StringVar(&cfg.userDSN, "user-dsn", envDefault("USER_MYSQL_DSN", "hyapp:hyapp@tcp(127.0.0.1:23306)/hyapp_user?parseTime=true&charset=utf8mb4&loc=UTC&multiStatements=true"), "user mysql dsn")
	flag.StringVar(&cfg.walletDSN, "wallet-dsn", envDefault("WALLET_MYSQL_DSN", "hyapp:hyapp@tcp(127.0.0.1:23306)/hyapp_wallet?parseTime=true&charset=utf8mb4&loc=UTC&multiStatements=true"), "wallet mysql dsn")
	flag.StringVar(&cfg.userAddr, "user-addr", envDefault("USER_GRPC_ADDR", "127.0.0.1:13005"), "user-service grpc address")
	flag.StringVar(&cfg.walletAddr, "wallet-addr", envDefault("WALLET_GRPC_ADDR", "127.0.0.1:13004"), "wallet-service grpc address")
	flag.StringVar(&cfg.roomAddr, "room-addr", envDefault("ROOM_GRPC_ADDR", "127.0.0.1:13001"), "room-service grpc address")
	flag.StringVar(&cfg.gameAddr, "game-addr", envDefault("GAME_GRPC_ADDR", "127.0.0.1:13008"), "game-service grpc address")
	flag.Parse()

	if strings.TrimSpace(cfg.runID) == "" {
		cfg.runID = fmt.Sprintf("%d", time.Now().UnixNano())
	}
	cfg.runID = sanitizeID(cfg.runID)
	cfg.roomID = shortID("dbroom", cfg.runID, 44)
	cfg.normalGift = shortID("dbg_normal", cfg.runID, 60)
	cfg.luckyGift = shortID("dbg_lucky", cfg.runID, 60)
	cfg.superGift = shortID("dbg_super", cfg.runID, 60)
	cfg.luckyPool = "lucky"
	cfg.superPool = "super_lucky"
	cfg.normalPrice = 200
	cfg.luckyPrice = 120
	cfg.superPrice = 160

	ctx, cancel := context.WithTimeout(context.Background(), 8*time.Minute)
	defer cancel()

	userDB, err := openDB(ctx, cfg.userDSN)
	if err != nil {
		log.Fatalf("open user db: %v", err)
	}
	defer userDB.Close()
	walletDB, err := openDB(ctx, cfg.walletDSN)
	if err != nil {
		log.Fatalf("open wallet db: %v", err)
	}
	defer walletDB.Close()

	userConn := mustDial(ctx, cfg.userAddr)
	defer userConn.Close()
	walletConn := mustDial(ctx, cfg.walletAddr)
	defer walletConn.Close()
	roomConn := mustDial(ctx, cfg.roomAddr)
	defer roomConn.Close()
	gameConn := mustDial(ctx, cfg.gameAddr)
	defer gameConn.Close()

	r := &runner{
		cfg:          cfg,
		userDB:       userDB,
		walletDB:     walletDB,
		authClient:   userv1.NewAuthServiceClient(userConn),
		hostClient:   userv1.NewUserHostAdminServiceClient(userConn),
		walletClient: walletv1.NewWalletServiceClient(walletConn),
		cronClient:   walletv1.NewWalletCronServiceClient(walletConn),
		roomClient:   roomv1.NewRoomCommandServiceClient(roomConn),
		gameAdmin:    gamev1.NewGameAdminServiceClient(gameConn),
		gameApp:      gamev1.NewGameAppServiceClient(gameConn),
	}
	if err := r.run(ctx); err != nil {
		log.Fatalf("databi real flow failed: %v", err)
	}
}

func (r *runner) run(ctx context.Context) error {
	if err := r.ensureUserMasterData(ctx); err != nil {
		return err
	}
	if err := r.ensureSalaryConfig(ctx); err != nil {
		return err
	}
	if err := r.ensureGameConfig(ctx); err != nil {
		return err
	}
	if err := r.createAccounts(ctx); err != nil {
		return err
	}
	if err := r.createBusinessRoles(ctx); err != nil {
		return err
	}
	if err := r.prepareWalletAndGifts(ctx); err != nil {
		return err
	}
	if err := r.simulateRechargeEvents(ctx); err != nil {
		return err
	}
	if err := r.runRoomFlow(ctx); err != nil {
		return err
	}
	if err := r.runSalaryFlow(ctx); err != nil {
		return err
	}
	if err := r.runGameFlow(ctx); err != nil {
		return err
	}
	if err := r.waitForAsyncSources(ctx); err != nil {
		return err
	}
	fmt.Printf("databi real business flow completed run_id=%s app=%s country_id=%d room_id=%s\n", r.cfg.runID, r.cfg.appCode, r.cfg.countryID, r.cfg.roomID)
	return nil
}

func (r *runner) ensureUserMasterData(ctx context.Context) error {
	now := time.Now().UnixMilli()
	_, err := r.userDB.ExecContext(ctx, `
		INSERT INTO countries (
			app_code, country_name, country_code, iso_alpha3, iso_numeric, country_display_name,
			phone_country_code, flag, enabled, sort_order, created_by_user_id, updated_by_user_id, created_at_ms, updated_at_ms
		) VALUES (?, 'United States', ?, 'USA', '840', 'United States', '+1', 'US', 1, 900, 0, 0, ?, ?)
		ON DUPLICATE KEY UPDATE
			country_name = VALUES(country_name),
			country_display_name = VALUES(country_display_name),
			enabled = 1,
			updated_at_ms = VALUES(updated_at_ms)`,
		r.cfg.appCode, countryCode, now, now)
	if err != nil {
		return fmt.Errorf("seed country: %w", err)
	}
	if err := r.userDB.QueryRowContext(ctx,
		`SELECT country_id FROM countries WHERE app_code = ? AND country_code = ?`,
		r.cfg.appCode, countryCode,
	).Scan(&r.cfg.countryID); err != nil {
		return fmt.Errorf("read country id: %w", err)
	}
	_, err = r.userDB.ExecContext(ctx, `
		INSERT INTO region_countries (
			app_code, region_id, country_code, status, created_by_user_id, updated_by_user_id, created_at_ms, updated_at_ms
		) VALUES (?, ?, ?, 'active', 0, 0, ?, ?)
		ON DUPLICATE KEY UPDATE
			region_id = VALUES(region_id),
			status = 'active',
			updated_at_ms = VALUES(updated_at_ms)`,
		r.cfg.appCode, regionID, countryCode, now, now)
	if err != nil {
		return fmt.Errorf("seed region country: %w", err)
	}
	fmt.Printf("seeded user master data country=%s country_id=%d region_id=%d\n", countryCode, r.cfg.countryID, regionID)
	return nil
}

func (r *runner) ensureSalaryConfig(ctx context.Context) error {
	now := time.Now().UnixMilli()
	_, err := r.walletDB.ExecContext(ctx, `
		INSERT INTO host_agency_salary_policies (
			app_code, policy_id, name, region_id, status, settlement_mode, settlement_trigger_mode,
			gift_coin_to_diamond_ratio, residual_diamond_to_usd_rate, effective_from_ms, effective_to_ms, created_at_ms, updated_at_ms
		) VALUES (?, 990001, 'Databi Local Daily Policy', ?, 'active', 'daily', 'automatic', 1.000000, 0.000000000000, 0, 0, ?, ?)
		ON DUPLICATE KEY UPDATE
			region_id = VALUES(region_id),
			status = 'active',
			settlement_mode = 'daily',
			settlement_trigger_mode = 'automatic',
			gift_coin_to_diamond_ratio = VALUES(gift_coin_to_diamond_ratio),
			updated_at_ms = VALUES(updated_at_ms)`,
		r.cfg.appCode, regionID, now, now)
	if err != nil {
		return fmt.Errorf("seed salary policy: %w", err)
	}
	_, err = r.walletDB.ExecContext(ctx, `
		INSERT INTO host_agency_salary_policy_levels (
			app_code, policy_id, level_no, required_diamonds, host_salary_usd_minor, host_coin_reward,
			agency_salary_usd_minor, status, sort_order, created_at_ms, updated_at_ms
		) VALUES (?, 990001, 1, 100, 150, 0, 50, 'active', 1, ?, ?)
		ON DUPLICATE KEY UPDATE
			required_diamonds = VALUES(required_diamonds),
			host_salary_usd_minor = VALUES(host_salary_usd_minor),
			agency_salary_usd_minor = VALUES(agency_salary_usd_minor),
			status = 'active',
			updated_at_ms = VALUES(updated_at_ms)`,
		r.cfg.appCode, now, now)
	if err != nil {
		return fmt.Errorf("seed salary policy level: %w", err)
	}
	_, err = r.walletDB.ExecContext(ctx,
		`DELETE FROM coin_seller_salary_exchange_rate_tiers WHERE app_code = ? AND region_id = ? AND sort_order = 990001`,
		r.cfg.appCode, regionID)
	if err != nil {
		return fmt.Errorf("clear salary exchange tier: %w", err)
	}
	_, err = r.walletDB.ExecContext(ctx, `
		INSERT INTO coin_seller_salary_exchange_rate_tiers (
			app_code, region_id, min_usd_minor, max_usd_minor, coin_per_usd, status,
			sort_order, created_by_user_id, updated_by_user_id, created_at_ms, updated_at_ms
		) VALUES (?, ?, 1, 99999999, 1000, 'active', 990001, 0, 0, ?, ?)`,
		r.cfg.appCode, regionID, now, now)
	if err != nil {
		return fmt.Errorf("seed salary exchange tier: %w", err)
	}
	fmt.Println("seeded wallet salary policy and salary-to-seller exchange tier")
	return nil
}

func (r *runner) ensureGameConfig(ctx context.Context) error {
	meta := r.gameMeta("game-admin")
	if _, err := r.gameAdmin.UpsertPlatform(ctx, &gamev1.UpsertPlatformRequest{
		Meta: meta,
		Platform: &gamev1.GamePlatform{
			AppCode:             r.cfg.appCode,
			PlatformCode:        "dice",
			PlatformName:        "Databi Dice",
			Status:              "active",
			ApiBaseUrl:          "https://local.hyapp.invalid/dice",
			SortOrder:           1,
			AdapterType:         "yomi_v4",
			CallbackSecret:      "local-databi-secret",
			CallbackIpWhitelist: []string{},
			AdapterConfigJson:   "{}",
		},
	}); err != nil {
		return fmt.Errorf("upsert game platform: %w", err)
	}
	if _, err := r.gameAdmin.UpsertCatalog(ctx, &gamev1.UpsertCatalogRequest{
		Meta: meta,
		Game: &gamev1.GameCatalogItem{
			AppCode:        r.cfg.appCode,
			GameId:         gameID,
			PlatformCode:   "dice",
			ProviderGameId: "dice",
			GameName:       "Databi Dice",
			Category:       "self",
			IconUrl:        "https://local.hyapp.invalid/dice.png",
			CoverUrl:       "https://local.hyapp.invalid/dice-cover.png",
			LaunchMode:     "h5_popup",
			Orientation:    "portrait",
			MinCoin:        100,
			Status:         "active",
			SortOrder:      1,
			Tags:           []string{"self"},
		},
	}); err != nil {
		return fmt.Errorf("upsert game catalog: %w", err)
	}
	if _, err := r.gameAdmin.UpdateDiceConfig(ctx, &gamev1.UpdateDiceConfigRequest{
		Meta: meta,
		Config: &gamev1.DiceConfig{
			AppCode: r.cfg.appCode,
			GameId:  gameID,
			Status:  "active",
			StakeOptions: []*gamev1.DiceStakeOption{{
				StakeCoin: 100,
				Enabled:   true,
				SortOrder: 1,
			}},
			FeeBps:           500,
			PoolBps:          100,
			MinPlayers:       2,
			MaxPlayers:       2,
			RobotEnabled:     false,
			RobotMatchWaitMs: 3000,
		},
	}); err != nil {
		return fmt.Errorf("upsert dice config: %w", err)
	}
	fmt.Println("seeded game platform/catalog/dice config through game-service RPC")
	return nil
}

func (r *runner) createAccounts(ctx context.Context) error {
	admin, err := r.quickCreate(ctx, "admin")
	if err != nil {
		return err
	}
	r.cfg.admin = admin
	fmt.Printf("created admin user_id=%d display=%s\n", admin.userID, admin.displayUserID)

	for _, spec := range []struct {
		key    string
		target *account
	}{
		{key: "sender", target: &r.cfg.sender},
		{key: "host", target: &r.cfg.host},
		{key: "seller", target: &r.cfg.seller},
		{key: "game_player", target: &r.cfg.gamePlayer},
		{key: "game_rival", target: &r.cfg.gameRival},
	} {
		acc, err := r.registerAppAccount(ctx, spec.key)
		if err != nil {
			return err
		}
		*spec.target = acc
		fmt.Printf("registered app user %-11s user_id=%d display=%s\n", spec.key, acc.userID, acc.displayUserID)
	}
	return nil
}

func (r *runner) quickCreate(ctx context.Context, role string) (account, error) {
	username := shortID("databi_"+role, r.cfg.runID, 28)
	deviceID := shortID("device_"+role, r.cfg.runID, 60)
	resp, err := r.authClient.QuickCreateAccount(ctx, &userv1.QuickCreateAccountRequest{
		Meta: &userv1.RequestMeta{
			RequestId:   r.commandID("quick_create_" + role),
			Caller:      "databi-real-flow",
			SentAtMs:    time.Now().UnixMilli(),
			DeviceId:    deviceID,
			ClientIp:    "127.0.0.1",
			UserAgent:   "databi-real-flow",
			CountryByIp: countryCode,
			AppCode:     r.cfg.appCode,
			Platform:    "ios",
			Language:    "en",
			Timezone:    "Asia/Shanghai",
		},
		Password:       "Databi@123456",
		Username:       username,
		Avatar:         "https://local.hyapp.invalid/avatar.png",
		Gender:         "female",
		Country:        countryCode,
		DeviceId:       deviceID,
		Device:         "local",
		OsVersion:      "17.0",
		Birth:          "1995-01-01",
		AppVersion:     "1.0.0",
		BuildNumber:    "1",
		Source:         "databi_real_flow",
		InstallChannel: "local",
		Platform:       "ios",
		Language:       "en",
		Timezone:       "Asia/Shanghai",
	})
	if err != nil {
		return account{}, fmt.Errorf("quick create %s: %w", role, err)
	}
	token := resp.GetToken()
	if token.GetUserId() <= 0 {
		return account{}, fmt.Errorf("quick create %s returned empty user id", role)
	}
	return account{userID: token.GetUserId(), displayUserID: token.GetDisplayUserId(), username: username}, nil
}

func (r *runner) registerAppAccount(ctx context.Context, role string) (account, error) {
	username := shortID("databi_"+role, r.cfg.runID, 28)
	deviceID := shortID("device_"+role, r.cfg.runID, 60)
	resp, err := r.authClient.RegisterThirdParty(ctx, &userv1.RegisterThirdPartyRequest{
		Meta: &userv1.RequestMeta{
			RequestId:   r.commandID("register_third_party_" + role),
			Caller:      "databi-real-flow",
			SentAtMs:    time.Now().UnixMilli(),
			DeviceId:    deviceID,
			ClientIp:    "127.0.0.1",
			UserAgent:   "databi-real-flow",
			CountryByIp: countryCode,
			AppCode:     r.cfg.appCode,
			Platform:    "ios",
			Language:    "en",
			Timezone:    "Asia/Shanghai",
		},
		Provider:       "wechat",
		Credential:     "openid_" + role + "_" + r.cfg.runID,
		DeviceId:       deviceID,
		Username:       username,
		Avatar:         "https://local.hyapp.invalid/avatar.png",
		Gender:         "female",
		Country:        countryCode,
		Device:         "local",
		OsVersion:      "17.0",
		Birth:          "1995-01-01",
		AppVersion:     "1.0.0",
		BuildNumber:    "1",
		Source:         "databi_app_registration",
		InstallChannel: "local",
		Platform:       "ios",
		Language:       "en",
		Timezone:       "Asia/Shanghai",
	})
	if err != nil {
		return account{}, fmt.Errorf("register app user %s: %w", role, err)
	}
	token := resp.GetToken()
	if token.GetUserId() <= 0 {
		return account{}, fmt.Errorf("register app user %s returned empty user id", role)
	}
	if !resp.GetIsNewUser() {
		return account{}, fmt.Errorf("register app user %s reused an existing third-party credential", role)
	}
	return account{userID: token.GetUserId(), displayUserID: token.GetDisplayUserId(), username: username}, nil
}

func (r *runner) createBusinessRoles(ctx context.Context) error {
	if _, err := r.hostClient.CreateCoinSeller(ctx, &userv1.CreateCoinSellerRequest{
		Meta:         r.userMeta("create_coin_seller"),
		CommandId:    r.commandID("create_coin_seller"),
		AdminUserId:  r.cfg.admin.userID,
		TargetUserId: r.cfg.seller.userID,
		Reason:       "databi real flow",
	}); err != nil {
		return fmt.Errorf("create coin seller: %w", err)
	}
	if _, err := r.hostClient.CreateAgency(ctx, &userv1.CreateAgencyRequest{
		Meta:        r.userMeta("create_agency"),
		CommandId:   r.commandID("create_agency"),
		AdminUserId: r.cfg.admin.userID,
		OwnerUserId: r.cfg.host.userID,
		Name:        shortID("Databi Agency", r.cfg.runID, 80),
		JoinEnabled: true,
		Reason:      "databi real flow",
	}); err != nil {
		return fmt.Errorf("create agency/host: %w", err)
	}
	fmt.Printf("created roles seller_user_id=%d host_agency_owner_user_id=%d\n", r.cfg.seller.userID, r.cfg.host.userID)
	return nil
}

func (r *runner) prepareWalletAndGifts(ctx context.Context) error {
	credits := []struct {
		userID int64
		amount int64
		label  string
	}{
		{userID: r.cfg.sender.userID, amount: 1_000_000, label: "sender"},
		{userID: r.cfg.gamePlayer.userID, amount: 10_000, label: "game_player"},
		{userID: r.cfg.gameRival.userID, amount: 10_000, label: "game_rival"},
	}
	for _, credit := range credits {
		if _, err := r.walletClient.AdminCreditAsset(ctx, &walletv1.AdminCreditAssetRequest{
			CommandId:      r.commandID("manual_credit_" + credit.label),
			TargetUserId:   credit.userID,
			AssetType:      "COIN",
			Amount:         credit.amount,
			OperatorUserId: r.cfg.admin.userID,
			Reason:         "databi real flow manual grant",
			EvidenceRef:    "local-real-flow",
			AppCode:        r.cfg.appCode,
		}); err != nil {
			return fmt.Errorf("admin credit %s: %w", credit.label, err)
		}
	}
	if _, err := r.walletClient.AdminCreditCoinSellerStock(ctx, &walletv1.AdminCreditCoinSellerStockRequest{
		CommandId:        r.commandID("seller_stock"),
		SellerUserId:     r.cfg.seller.userID,
		StockType:        "usdt_purchase",
		CoinAmount:       500_000,
		PaidCurrencyCode: "USDT",
		PaidAmountMicro:  10_000_000,
		PaymentRef:       r.commandID("seller_stock_payment"),
		EvidenceRef:      "local-real-flow",
		OperatorUserId:   r.cfg.admin.userID,
		Reason:           "databi real flow seller stock",
		AppCode:          r.cfg.appCode,
		SellerCountryId:  r.cfg.countryID,
		SellerRegionId:   regionID,
	}); err != nil {
		return fmt.Errorf("credit seller stock: %w", err)
	}
	if _, err := r.walletClient.TransferCoinFromSeller(ctx, &walletv1.TransferCoinFromSellerRequest{
		CommandId:       r.commandID("seller_transfer"),
		SellerUserId:    r.cfg.seller.userID,
		TargetUserId:    r.cfg.sender.userID,
		Amount:          50_000,
		Reason:          "databi real flow seller transfer",
		AppCode:         r.cfg.appCode,
		SellerRegionId:  regionID,
		TargetRegionId:  regionID,
		TargetCountryId: r.cfg.countryID,
	}); err != nil {
		return fmt.Errorf("seller transfer coin: %w", err)
	}
	if _, err := r.walletClient.CreditTaskReward(ctx, &walletv1.CreditTaskRewardRequest{
		CommandId:    r.commandID("task_reward"),
		AppCode:      r.cfg.appCode,
		TargetUserId: r.cfg.sender.userID,
		Amount:       1_000,
		TaskType:     "daily",
		TaskId:       "databi_daily",
		CycleKey:     time.Now().UTC().Format("2006-01-02"),
		Reason:       "databi real flow platform reward",
	}); err != nil {
		return fmt.Errorf("task reward: %w", err)
	}
	if _, err := r.walletClient.CreditLuckyGiftReward(ctx, &walletv1.CreditLuckyGiftRewardRequest{
		CommandId:       r.commandID("super_lucky_reward"),
		AppCode:         r.cfg.appCode,
		TargetUserId:    r.cfg.sender.userID,
		Amount:          2_000,
		DrawId:          r.commandID("super_lucky_draw"),
		RoomId:          r.cfg.roomID,
		GiftId:          r.cfg.superGift,
		PoolId:          r.cfg.superPool,
		Reason:          "databi real flow super lucky payout",
		VisibleRegionId: regionID,
		CountryId:       r.cfg.countryID,
	}); err != nil {
		return fmt.Errorf("lucky gift reward: %w", err)
	}
	if err := r.createGift(ctx, r.cfg.normalGift, "normal", r.cfg.normalPrice); err != nil {
		return err
	}
	if err := r.createGift(ctx, r.cfg.luckyGift, "lucky", r.cfg.luckyPrice); err != nil {
		return err
	}
	if err := r.createGift(ctx, r.cfg.superGift, "super_lucky", r.cfg.superPrice); err != nil {
		return err
	}
	fmt.Println("prepared wallet balances, seller stock/transfer, platform reward and gift configs")
	return nil
}

func (r *runner) createGift(ctx context.Context, giftID string, giftType string, price int64) error {
	if err := r.ensureGiftType(ctx, giftType); err != nil {
		return err
	}
	mgrGrant := true
	resourceResp, err := r.walletClient.CreateResource(ctx, &walletv1.CreateResourceRequest{
		RequestId:           r.commandID("resource_" + giftID),
		AppCode:             r.cfg.appCode,
		ResourceCode:        "res_" + giftID,
		ResourceType:        "gift",
		Name:                "Databi " + giftType,
		Status:              "active",
		Grantable:           true,
		GrantStrategy:       "increase_quantity",
		UsageScopes:         []string{"gift"},
		AssetUrl:            "https://local.hyapp.invalid/" + giftID + ".png",
		PreviewUrl:          "https://local.hyapp.invalid/" + giftID + "-preview.png",
		AnimationUrl:        "https://local.hyapp.invalid/" + giftID + ".svga",
		MetadataJson:        "{}",
		SortOrder:           900,
		OperatorUserId:      r.cfg.admin.userID,
		ManagerGrantEnabled: &mgrGrant,
		PriceType:           "coin",
		CoinPrice:           price,
	})
	if err != nil {
		return fmt.Errorf("create resource %s: %w", giftID, err)
	}
	if _, err := r.walletClient.CreateGiftConfig(ctx, &walletv1.CreateGiftConfigRequest{
		RequestId:        r.commandID("gift_config_" + giftID),
		AppCode:          r.cfg.appCode,
		GiftId:           giftID,
		ResourceId:       resourceResp.GetResource().GetResourceId(),
		Status:           "active",
		Name:             "Databi " + giftType,
		SortOrder:        900,
		PresentationJson: "{}",
		PriceVersion:     "v1",
		CoinPrice:        price,
		HeatValue:        price,
		EffectiveAtMs:    time.Now().UnixMilli(),
		OperatorUserId:   r.cfg.admin.userID,
		RegionIds:        []int64{regionID},
		GiftTypeCode:     giftType,
		ChargeAssetType:  "COIN",
		EffectiveFromMs:  0,
		EffectiveToMs:    0,
	}); err != nil {
		return fmt.Errorf("create gift config %s: %w", giftID, err)
	}
	return nil
}

func (r *runner) ensureGiftType(ctx context.Context, giftType string) error {
	var count int64
	if err := r.walletDB.QueryRowContext(ctx,
		`SELECT COUNT(*) FROM gift_type_configs WHERE app_code = ? AND type_code = ?`,
		r.cfg.appCode, giftType,
	).Scan(&count); err != nil {
		return fmt.Errorf("read gift type %s: %w", giftType, err)
	}
	if count > 0 {
		return nil
	}
	if _, err := r.walletClient.UpsertGiftTypeConfig(ctx, &walletv1.UpsertGiftTypeConfigRequest{
		RequestId:      r.commandID("gift_type_" + giftType),
		AppCode:        r.cfg.appCode,
		TypeCode:       giftType,
		Name:           giftType,
		TabKey:         shortID("databi_"+giftType, r.cfg.runID, 32),
		Status:         "active",
		SortOrder:      900,
		OperatorUserId: r.cfg.admin.userID,
	}); err != nil {
		return fmt.Errorf("upsert gift type %s: %w", giftType, err)
	}
	return nil
}

func (r *runner) simulateRechargeEvents(ctx context.Context) error {
	if err := r.insertSimulatedRecharge(ctx, "google", 199, 1_990_000); err != nil {
		return err
	}
	if err := r.insertSimulatedRecharge(ctx, "mifapay", 299, 2_990_000); err != nil {
		return err
	}
	fmt.Println("inserted simulated google/mifapay recharge outbox events")
	return nil
}

func (r *runner) insertSimulatedRecharge(ctx context.Context, channel string, usdMinor int64, amountMicro int64) error {
	now := time.Now().UnixMilli()
	eventID := r.commandID("sim_recharge_" + channel)
	payload := map[string]any{
		"app_code":               r.cfg.appCode,
		"provider":               channel,
		"channel":                channel,
		"recharge_type":          channel,
		"target_user_id":         r.cfg.sender.userID,
		"target_country_id":      r.cfg.countryID,
		"country_id":             r.cfg.countryID,
		"target_region_id":       regionID,
		"region_id":              regionID,
		"amount_micro":           amountMicro,
		"recharge_usd_minor":     usdMinor,
		"recharge_currency_code": "USD",
		"recharge_sequence":      1,
		"created_at_ms":          now,
	}
	raw, err := json.Marshal(payload)
	if err != nil {
		return err
	}
	_, err = r.walletDB.ExecContext(ctx, `
		INSERT INTO wallet_outbox (
			app_code, event_id, event_type, transaction_id, command_id, user_id, asset_type,
			available_delta, frozen_delta, payload, status, worker_id, lock_until_ms, retry_count,
			next_retry_at_ms, last_error, created_at_ms, updated_at_ms
		) VALUES (?, ?, 'WalletRechargeRecorded', ?, ?, ?, 'COIN', 0, 0, CAST(? AS JSON), 'pending', '', NULL, 0, NULL, NULL, ?, ?)
		ON DUPLICATE KEY UPDATE payload = VALUES(payload), updated_at_ms = VALUES(updated_at_ms)`,
		r.cfg.appCode, eventID, eventID, eventID, r.cfg.sender.userID, string(raw), now, now)
	if err != nil {
		return fmt.Errorf("insert simulated %s recharge: %w", channel, err)
	}
	return nil
}

func (r *runner) runRoomFlow(ctx context.Context) error {
	if _, err := r.roomClient.CreateRoom(ctx, &roomv1.CreateRoomRequest{
		Meta:             r.roomMeta("create_room", r.cfg.sender.userID),
		Mode:             "voice",
		VisibleRegionId:  regionID,
		RoomName:         "Databi Real Flow",
		RoomAvatar:       "https://local.hyapp.invalid/room.png",
		RoomDescription:  "databi real flow",
		RoomShortId:      firstNonEmpty(r.cfg.sender.displayUserID, fmt.Sprintf("%d", r.cfg.sender.userID)),
		OwnerCountryCode: countryCode,
	}); err != nil {
		return fmt.Errorf("create room: %w", err)
	}
	for _, acc := range []account{r.cfg.host, r.cfg.sender} {
		if _, err := r.roomClient.JoinRoom(ctx, &roomv1.JoinRoomRequest{
			Meta:           r.roomMeta("join_"+acc.username, acc.userID),
			Role:           "member",
			ActorCountryId: r.cfg.countryID,
			ActorRegionId:  regionID,
			ActorDisplayProfile: &roomv1.RoomUserDisplayProfile{
				UserId:   acc.userID,
				Nickname: acc.username,
				Avatar:   "https://local.hyapp.invalid/avatar.png",
			},
		}); err != nil {
			return fmt.Errorf("join room user %d: %w", acc.userID, err)
		}
	}
	micResp, err := r.roomClient.MicUp(ctx, &roomv1.MicUpRequest{
		Meta:   r.roomMeta("mic_up", r.cfg.host.userID),
		SeatNo: 1,
		TargetDisplayProfile: &roomv1.RoomUserDisplayProfile{
			UserId:   r.cfg.host.userID,
			Nickname: r.cfg.host.username,
			Avatar:   "https://local.hyapp.invalid/avatar.png",
		},
	})
	if err != nil {
		return fmt.Errorf("mic up: %w", err)
	}
	if _, err := r.roomClient.ConfirmMicPublishing(ctx, &roomv1.ConfirmMicPublishingRequest{
		Meta:         r.roomMeta("mic_confirm", r.cfg.host.userID),
		TargetUserId: r.cfg.host.userID,
		MicSessionId: micResp.GetMicSessionId(),
		RoomVersion:  micResp.GetResult().GetRoomVersion(),
		EventTimeMs:  time.Now().UnixMilli(),
		Source:       "databi-real-flow",
	}); err != nil {
		return fmt.Errorf("confirm mic publishing: %w", err)
	}
	time.Sleep(2200 * time.Millisecond)
	if _, err := r.roomClient.MicHeartbeat(ctx, &roomv1.MicHeartbeatRequest{
		Meta:         r.roomMeta("mic_heartbeat", r.cfg.host.userID),
		TargetUserId: r.cfg.host.userID,
		MicSessionId: micResp.GetMicSessionId(),
	}); err != nil {
		return fmt.Errorf("mic heartbeat: %w", err)
	}
	if _, err := r.roomClient.MicDown(ctx, &roomv1.MicDownRequest{
		Meta:         r.roomMeta("mic_down", r.cfg.host.userID),
		TargetUserId: r.cfg.host.userID,
		Reason:       "databi-real-flow",
	}); err != nil {
		return fmt.Errorf("mic down: %w", err)
	}
	for _, gift := range []struct {
		id     string
		count  int32
		poolID string
	}{
		{id: r.cfg.normalGift, count: 1},
		{id: r.cfg.luckyGift, count: 1, poolID: r.cfg.luckyPool},
		{id: r.cfg.superGift, count: 1, poolID: r.cfg.superPool},
	} {
		if _, err := r.roomClient.SendGift(ctx, &roomv1.SendGiftRequest{
			Meta:                    r.roomMeta("send_gift_"+gift.id, r.cfg.sender.userID),
			TargetUserId:            r.cfg.host.userID,
			GiftId:                  gift.id,
			GiftCount:               gift.count,
			TargetType:              "user",
			PoolId:                  gift.poolID,
			TargetIsHost:            true,
			TargetHostRegionId:      regionID,
			TargetAgencyOwnerUserId: r.cfg.host.userID,
			SenderRegionId:          regionID,
			SenderCountryId:         r.cfg.countryID,
			SenderDisplayProfile: &roomv1.SendGiftDisplayProfile{
				UserId:   r.cfg.sender.userID,
				Username: r.cfg.sender.username,
				Avatar:   "https://local.hyapp.invalid/avatar.png",
			},
			TargetDisplayProfiles: []*roomv1.SendGiftDisplayProfile{{
				UserId:   r.cfg.host.userID,
				Username: r.cfg.host.username,
				Avatar:   "https://local.hyapp.invalid/avatar.png",
			}},
		}); err != nil {
			return fmt.Errorf("send gift %s: %w", gift.id, err)
		}
	}
	fmt.Println("completed room join, mic, normal/lucky/super-lucky gift flows")
	return nil
}

func (r *runner) runSalaryFlow(ctx context.Context) error {
	cycleKey := time.Now().UTC().Format("2006-01")
	resp, err := r.cronClient.ProcessHostSalaryDailySettlementBatch(ctx, &walletv1.CronBatchRequest{
		RequestId:      r.commandID("salary_cron_request"),
		AppCode:        r.cfg.appCode,
		RunId:          r.commandID("salary_cron"),
		WorkerId:       "databi-real-flow",
		BatchSize:      10,
		LockTtlMs:      30_000,
		TriggerMode:    "automatic",
		CycleKey:       cycleKey,
		SettlementRole: "",
	})
	if err != nil {
		return fmt.Errorf("salary daily settlement cron: %w", err)
	}
	fmt.Printf("salary settlement cron success=%d failure=%d\n", resp.GetSuccessCount(), resp.GetFailureCount())
	if _, err := r.walletClient.TransferSalaryToCoinSeller(ctx, &walletv1.TransferSalaryToCoinSellerRequest{
		CommandId:       r.commandID("salary_to_seller"),
		SourceUserId:    r.cfg.host.userID,
		SellerUserId:    r.cfg.seller.userID,
		SalaryAssetType: "HOST_SALARY_USD",
		SalaryUsdMinor:  100,
		RegionId:        regionID,
		Reason:          "databi real flow salary to seller",
		AppCode:         r.cfg.appCode,
	}); err != nil {
		return fmt.Errorf("transfer salary to seller: %w", err)
	}
	fmt.Println("completed host/agency salary settlement and salary-to-seller transfer")
	return nil
}

func (r *runner) runGameFlow(ctx context.Context) error {
	createResp, err := r.gameApp.CreateDiceMatch(ctx, &gamev1.CreateDiceMatchRequest{
		Meta:       r.gameMeta("dice_create"),
		UserId:     r.cfg.gamePlayer.userID,
		GameId:     gameID,
		RoomId:     r.cfg.roomID,
		RegionId:   regionID,
		StakeCoin:  100,
		MinPlayers: 2,
		MaxPlayers: 2,
		CountryId:  r.cfg.countryID,
	})
	if err != nil {
		return fmt.Errorf("create dice match: %w", err)
	}
	matchID := createResp.GetMatch().GetMatchId()
	if _, err := r.gameApp.JoinDiceMatch(ctx, &gamev1.JoinDiceMatchRequest{
		Meta:    r.gameMeta("dice_join"),
		UserId:  r.cfg.gameRival.userID,
		MatchId: matchID,
		GameId:  gameID,
	}); err != nil {
		return fmt.Errorf("join dice match: %w", err)
	}
	for attempt := 1; attempt <= 5; attempt++ {
		rollResp, err := r.gameApp.RollDiceMatch(ctx, &gamev1.RollDiceMatchRequest{
			Meta:    r.gameMeta(fmt.Sprintf("dice_roll_%d", attempt)),
			UserId:  r.cfg.gamePlayer.userID,
			MatchId: matchID,
			GameId:  gameID,
		})
		if err != nil {
			return fmt.Errorf("roll dice match attempt %d: %w", attempt, err)
		}
		status := rollResp.GetMatch().GetStatus()
		fmt.Printf("dice roll attempt=%d status=%s result=%s\n", attempt, status, rollResp.GetMatch().GetResult())
		if status == "settled" {
			fmt.Println("completed game dice settlement")
			return nil
		}
		time.Sleep(2300 * time.Millisecond)
	}
	return fmt.Errorf("dice match %s did not settle after retries", matchID)
}

func (r *runner) waitForAsyncSources(ctx context.Context) error {
	checks := []struct {
		name  string
		db    *sql.DB
		query string
		args  []any
	}{
		{
			name:  "mic daily stats",
			db:    r.userDB,
			query: `SELECT COUNT(*) FROM user_outbox WHERE app_code = ? AND event_type = 'UserMicDailyStatsUpdated' AND created_at_ms >= ?`,
			args:  []any{r.cfg.appCode, runStartMS(r.cfg.runID)},
		},
		{
			name:  "game settled outbox",
			db:    nil,
			query: "",
		},
	}
	for _, check := range checks {
		if check.db == nil {
			continue
		}
		if err := waitForCount(ctx, check.db, check.query, check.args...); err != nil {
			return fmt.Errorf("wait %s: %w", check.name, err)
		}
	}
	time.Sleep(1500 * time.Millisecond)
	return nil
}

func (r *runner) userMeta(request string) *userv1.RequestMeta {
	return &userv1.RequestMeta{
		RequestId: r.commandID(request),
		Caller:    "databi-real-flow",
		SentAtMs:  time.Now().UnixMilli(),
		AppCode:   r.cfg.appCode,
		Platform:  "ios",
		Language:  "en",
		Timezone:  "Asia/Shanghai",
	}
}

func (r *runner) roomMeta(action string, actorUserID int64) *roomv1.RequestMeta {
	return &roomv1.RequestMeta{
		RequestId:     r.commandID(action + "_request"),
		CommandId:     r.commandID(action),
		ActorUserId:   actorUserID,
		RoomId:        r.cfg.roomID,
		GatewayNodeId: "databi-real-flow",
		SentAtMs:      time.Now().UnixMilli(),
		AppCode:       r.cfg.appCode,
	}
}

func (r *runner) gameMeta(request string) *gamev1.RequestMeta {
	return &gamev1.RequestMeta{
		RequestId:     r.commandID(request),
		Caller:        "databi-real-flow",
		GatewayNodeId: "databi-real-flow",
		ActorUserId:   r.cfg.admin.userID,
		SentAtMs:      time.Now().UnixMilli(),
		AppCode:       r.cfg.appCode,
	}
}

func (r *runner) commandID(action string) string {
	return shortID("databi_"+action, r.cfg.runID, 120)
}

func openDB(ctx context.Context, dsn string) (*sql.DB, error) {
	db, err := sql.Open("mysql", strings.TrimSpace(dsn))
	if err != nil {
		return nil, err
	}
	db.SetMaxOpenConns(4)
	db.SetMaxIdleConns(2)
	if err := db.PingContext(ctx); err != nil {
		_ = db.Close()
		return nil, err
	}
	return db, nil
}

func mustDial(ctx context.Context, addr string) *grpc.ClientConn {
	conn, err := grpc.DialContext(ctx, addr, grpc.WithTransportCredentials(insecure.NewCredentials()), grpc.WithBlock())
	if err != nil {
		log.Fatalf("dial %s: %v", addr, err)
	}
	return conn
}

func waitForCount(ctx context.Context, db *sql.DB, query string, args ...any) error {
	deadline := time.Now().Add(20 * time.Second)
	for {
		var count int64
		if err := db.QueryRowContext(ctx, query, args...).Scan(&count); err != nil {
			return err
		}
		if count > 0 {
			return nil
		}
		if time.Now().After(deadline) {
			return fmt.Errorf("timed out waiting for count > 0")
		}
		select {
		case <-ctx.Done():
			return ctx.Err()
		case <-time.After(500 * time.Millisecond):
		}
	}
}

func runStartMS(runID string) int64 {
	if value, err := time.Parse("20060102150405", runID); err == nil {
		return value.UnixMilli()
	}
	return time.Now().Add(-10 * time.Minute).UnixMilli()
}

func shortID(prefix string, suffix string, limit int) string {
	id := sanitizeID(prefix) + "_" + sanitizeID(suffix)
	if len(id) <= limit {
		return id
	}
	if limit <= 8 {
		return id[:limit]
	}
	tailLen := limit - len(sanitizeID(prefix)) - 1
	if tailLen <= 0 {
		return id[:limit]
	}
	if tailLen > len(sanitizeID(suffix)) {
		tailLen = len(sanitizeID(suffix))
	}
	return sanitizeID(prefix) + "_" + sanitizeID(suffix)[:tailLen]
}

func sanitizeID(value string) string {
	value = strings.TrimSpace(value)
	if value == "" {
		value = fmt.Sprintf("%d", time.Now().UnixNano())
	}
	var b strings.Builder
	for _, r := range value {
		switch {
		case r >= 'a' && r <= 'z':
			b.WriteRune(r)
		case r >= 'A' && r <= 'Z':
			b.WriteRune(r)
		case r >= '0' && r <= '9':
			b.WriteRune(r)
		case r == '_' || r == '-':
			b.WriteRune(r)
		default:
			b.WriteByte('_')
		}
	}
	return b.String()
}

func firstNonEmpty(values ...string) string {
	for _, value := range values {
		if strings.TrimSpace(value) != "" {
			return strings.TrimSpace(value)
		}
	}
	return ""
}

func envDefault(key string, fallback string) string {
	if value := strings.TrimSpace(os.Getenv(key)); value != "" {
		return value
	}
	return fallback
}