546 lines
21 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 mysqltest provides real wallet-service MySQL repositories for tests.
package mysqltest
import (
"context"
"database/sql"
"fmt"
"testing"
"time"
"hyapp/internal/testutil/mysqlschema"
"hyapp/services/wallet-service/internal/domain/ledger"
mysqlstorage "hyapp/services/wallet-service/internal/storage/mysql"
)
// Repository wraps the production MySQL wallet repository with test seed helpers.
type Repository struct {
*mysqlstorage.Repository
t testing.TB
schema *mysqlschema.Schema
}
type VipRewardResourceSeed struct {
ResourceID int64
ResourceCode string
ResourceType string
Name string
GrantStrategy string
SortOrder int
}
// CountRows returns a simple count from a known wallet table for persistence assertions.
func (r *Repository) CountRows(table string, where string, args ...any) int {
r.t.Helper()
query := "SELECT COUNT(*) FROM " + validateTableName(table)
if where != "" {
query += " WHERE " + where
}
var count int
if err := r.schema.DB.QueryRowContext(context.Background(), query, args...).Scan(&count); err != nil {
r.t.Fatalf("count %s rows failed: %v\nquery=%s", table, err, query)
}
return count
}
func (r *Repository) ActiveResourceEntitlement(userID int64, resourceID int64) (expiresAtMS int64, quantity int64, remainingQuantity int64) {
r.t.Helper()
err := r.schema.DB.QueryRowContext(context.Background(), `
SELECT expires_at_ms, quantity, remaining_quantity
FROM user_resource_entitlements
WHERE user_id = ? AND resource_id = ? AND status = 'active'
ORDER BY expires_at_ms DESC, created_at_ms DESC
LIMIT 1
`, userID, resourceID).Scan(&expiresAtMS, &quantity, &remainingQuantity)
if err != nil {
r.t.Fatalf("query active resource entitlement failed: %v", err)
}
return expiresAtMS, quantity, remainingQuantity
}
// HostSalarySettlementIDs returns the persisted settlement IDs for one host/cycle in creation order.
func (r *Repository) HostSalarySettlementIDs(userID int64, cycleKey string) []string {
r.t.Helper()
rows, err := r.schema.DB.QueryContext(context.Background(), `
SELECT settlement_id
FROM host_salary_settlement_records
WHERE user_id = ? AND cycle_key = ?
ORDER BY created_at_ms ASC, settlement_id ASC
`, userID, cycleKey)
if err != nil {
r.t.Fatalf("query host salary settlement ids failed: %v", err)
}
defer rows.Close()
var ids []string
for rows.Next() {
var id string
if err := rows.Scan(&id); err != nil {
r.t.Fatalf("scan host salary settlement id failed: %v", err)
}
ids = append(ids, id)
}
if err := rows.Err(); err != nil {
r.t.Fatalf("iterate host salary settlement ids failed: %v", err)
}
return ids
}
// MoveHostPeriodCycle merges a real gift-created current-cycle account into another cycle for month-boundary tests.
func (r *Repository) MoveHostPeriodCycle(userID int64, fromCycle string, toCycle string) {
r.t.Helper()
if fromCycle == "" || toCycle == "" || fromCycle == toCycle {
return
}
tx, err := r.schema.DB.BeginTx(context.Background(), nil)
if err != nil {
r.t.Fatalf("begin move host period cycle failed: %v", err)
}
defer func() { _ = tx.Rollback() }()
var sourceTotal int64
var sourceGiftTotal int64
var sourceLastGiftAt int64
var sourceUpdatedAt int64
err = tx.QueryRowContext(context.Background(), `
SELECT total_diamonds, gift_diamond_total, last_gift_at_ms, updated_at_ms
FROM host_period_diamond_accounts
WHERE user_id = ? AND cycle_key = ?
FOR UPDATE
`, userID, fromCycle).Scan(&sourceTotal, &sourceGiftTotal, &sourceLastGiftAt, &sourceUpdatedAt)
if err != nil {
r.t.Fatalf("read source host period cycle failed: %v", err)
}
var targetTotal int64
err = tx.QueryRowContext(context.Background(), `
SELECT total_diamonds
FROM host_period_diamond_accounts
WHERE user_id = ? AND cycle_key = ?
FOR UPDATE
`, userID, toCycle).Scan(&targetTotal)
if err != nil {
if err == sql.ErrNoRows {
if _, err := tx.ExecContext(context.Background(), `
UPDATE host_period_diamond_accounts
SET cycle_key = ?
WHERE user_id = ? AND cycle_key = ?
`, toCycle, userID, fromCycle); err != nil {
r.t.Fatalf("move first host period account failed: %v", err)
}
} else {
r.t.Fatalf("read target host period cycle failed: %v", err)
}
} else {
if _, err := tx.ExecContext(context.Background(), `
UPDATE host_period_diamond_accounts
SET total_diamonds = total_diamonds + ?,
gift_diamond_total = gift_diamond_total + ?,
version = version + 1,
last_gift_at_ms = ?,
updated_at_ms = ?
WHERE user_id = ? AND cycle_key = ?
`, sourceTotal, sourceGiftTotal, sourceLastGiftAt, sourceUpdatedAt, userID, toCycle); err != nil {
r.t.Fatalf("merge target host period account failed: %v", err)
}
if _, err := tx.ExecContext(context.Background(), `
DELETE FROM host_period_diamond_accounts
WHERE user_id = ? AND cycle_key = ?
`, userID, fromCycle); err != nil {
r.t.Fatalf("delete merged source host period account failed: %v", err)
}
}
if _, err := tx.ExecContext(context.Background(), `
UPDATE host_period_diamond_entries
SET cycle_key = ?
WHERE user_id = ? AND cycle_key = ?
`, toCycle, userID, fromCycle); err != nil {
r.t.Fatalf("move host period entries failed: %v", err)
}
if err := tx.Commit(); err != nil {
r.t.Fatalf("commit move host period cycle failed: %v", err)
}
}
// HostPeriodGiftDiamondTotal returns the current accumulated gift diamonds for a host.
func (r *Repository) HostPeriodGiftDiamondTotal(userID int64) int64 {
r.t.Helper()
var total int64
err := r.schema.DB.QueryRowContext(context.Background(), `
SELECT COALESCE(SUM(gift_diamond_total), 0)
FROM host_period_diamond_accounts
WHERE user_id = ?`, userID).Scan(&total)
if err != nil {
r.t.Fatalf("query host period gift diamond total failed: %v", err)
}
return total
}
// InsertRawGiftPrice allows tests to seed inactive or future prices without service helpers.
func (r *Repository) InsertRawGiftPrice(giftID string, priceVersion string, status string, effectiveAtMs int64) {
r.t.Helper()
nowMs := time.Now().UnixMilli()
_, err := r.schema.DB.ExecContext(context.Background(), `
INSERT INTO wallet_gift_prices (
gift_id, price_version, status, coin_price, gift_point_amount, heat_value,
effective_at_ms, created_at_ms, updated_at_ms
) VALUES (?, ?, ?, 1, 1, 1, ?, ?, ?)
`, giftID, priceVersion, status, effectiveAtMs, nowMs, nowMs)
if err != nil {
r.t.Fatalf("insert raw gift price failed: %v", err)
}
}
// NewRepository creates an isolated wallet schema and opens the production repository.
func NewRepository(t testing.TB) *Repository {
t.Helper()
schema := mysqlschema.New(t, mysqlschema.Config{
EnvVar: "WALLET_SERVICE_MYSQL_TEST_DSN",
InitDBPath: initDBPath(t),
DatabasePrefix: "hyapp_wallet_test",
})
repository, err := mysqlstorage.Open(context.Background(), schema.DSN)
if err != nil {
t.Fatalf("open wallet mysql repository failed: %v", err)
}
wrapped := &Repository{Repository: repository, t: t, schema: schema}
t.Cleanup(wrapped.Close)
return wrapped
}
// Close shuts down the repository; mysqlschema owns schema cleanup.
func (r *Repository) Close() {
r.t.Helper()
if r.Repository != nil {
_ = r.Repository.Close()
}
}
// SetBalance prepares a sender COIN account for debit tests.
func (r *Repository) SetBalance(userID int64, balance int64) {
r.SetAssetBalance(userID, "COIN", balance)
}
func (r *Repository) SeedVIPRewardGroup(level int32, groupID int64, resources []VipRewardResourceSeed) {
r.t.Helper()
nowMs := time.Now().UnixMilli()
_, err := r.schema.DB.ExecContext(context.Background(), `
INSERT INTO resource_groups (
group_id, group_code, name, status, description, sort_order,
created_by_user_id, updated_by_user_id, created_at_ms, updated_at_ms
) VALUES (?, ?, ?, 'active', '', 0, 0, 0, ?, ?)
ON DUPLICATE KEY UPDATE
status = VALUES(status),
updated_at_ms = VALUES(updated_at_ms)
`, groupID, fmt.Sprintf("vip_test_group_%d", groupID), fmt.Sprintf("VIP test group %d", groupID), nowMs, nowMs)
if err != nil {
r.t.Fatalf("seed vip reward group failed: %v", err)
}
if _, err := r.schema.DB.ExecContext(context.Background(), `DELETE FROM resource_group_items WHERE group_id = ?`, groupID); err != nil {
r.t.Fatalf("clear vip reward group items failed: %v", err)
}
for index, resource := range resources {
sortOrder := resource.SortOrder
if sortOrder == 0 {
sortOrder = (index + 1) * 10
}
name := resource.Name
if name == "" {
name = resource.ResourceCode
}
_, err := r.schema.DB.ExecContext(context.Background(), `
INSERT INTO resources (
resource_id, resource_code, resource_type, name, status,
grantable, manager_grant_enabled, grant_strategy, wallet_asset_type,
wallet_asset_amount, price_type, coin_price, gift_point_amount,
usage_scope_json, asset_url, preview_url, animation_url,
metadata_json, sort_order, created_by_user_id, updated_by_user_id,
created_at_ms, updated_at_ms
) VALUES (?, ?, ?, ?, 'active', TRUE, TRUE, ?, '', 0, '', 0, 0,
JSON_ARRAY(), '', '', '', NULL, ?, 0, 0, ?, ?)
ON DUPLICATE KEY UPDATE
resource_type = VALUES(resource_type),
name = VALUES(name),
status = VALUES(status),
grant_strategy = VALUES(grant_strategy),
sort_order = VALUES(sort_order),
updated_at_ms = VALUES(updated_at_ms)
`, resource.ResourceID, resource.ResourceCode, resource.ResourceType, name, resource.GrantStrategy, sortOrder, nowMs, nowMs)
if err != nil {
r.t.Fatalf("seed vip reward resource failed: %v", err)
}
_, err = r.schema.DB.ExecContext(context.Background(), `
INSERT INTO resource_group_items (
group_id, item_type, resource_id, wallet_asset_type, wallet_asset_amount,
quantity, duration_ms, sort_order, created_at_ms, updated_at_ms
) VALUES (?, 'resource', ?, '', 0, 1, 0, ?, ?, ?)
`, groupID, resource.ResourceID, sortOrder, nowMs, nowMs)
if err != nil {
r.t.Fatalf("seed vip reward group item failed: %v", err)
}
}
if _, err := r.schema.DB.ExecContext(context.Background(), `
UPDATE vip_levels
SET reward_resource_group_id = ?, updated_at_ms = ?
WHERE level = ?
`, groupID, nowMs, level); err != nil {
r.t.Fatalf("update vip level reward group failed: %v", err)
}
}
// SetAssetBalance prepares a sender account for debit tests.
func (r *Repository) SetAssetBalance(userID int64, assetType string, balance int64) {
r.t.Helper()
nowMs := time.Now().UnixMilli()
_, err := r.schema.DB.ExecContext(context.Background(), `
INSERT INTO wallet_accounts (user_id, asset_type, available_amount, frozen_amount, version, created_at_ms, updated_at_ms)
VALUES (?, ?, ?, 0, 1, ?, ?)
ON DUPLICATE KEY UPDATE
available_amount = VALUES(available_amount),
frozen_amount = VALUES(frozen_amount),
version = version + 1,
updated_at_ms = VALUES(updated_at_ms)
`, userID, assetType, balance, nowMs, nowMs)
if err != nil {
r.t.Fatalf("seed wallet balance failed: %v", err)
}
}
// SetGiftPrice overrides or inserts a gift settlement price for server-side billing tests.
func (r *Repository) SetGiftPrice(giftID string, priceVersion string, coinPrice int64, giftPointAmount int64, heatValue int64) {
r.t.Helper()
nowMs := time.Now().UnixMilli()
// DebitGift 真实链路先校验 active gift_config再读取 wallet_gift_prices测试 seed 必须同时补齐资源、礼物配置和区域关系。
result, err := r.schema.DB.ExecContext(context.Background(), `
INSERT INTO resources (
resource_code, resource_type, name, status, grantable, manager_grant_enabled,
grant_strategy, price_type, coin_price, gift_point_amount, usage_scope_json,
created_at_ms, updated_at_ms
) VALUES (?, 'gift', ?, 'active', TRUE, TRUE, 'new_entitlement', 'coin', ?, ?, '[]', ?, ?)
ON DUPLICATE KEY UPDATE
resource_id = LAST_INSERT_ID(resource_id),
status = VALUES(status),
coin_price = VALUES(coin_price),
gift_point_amount = VALUES(gift_point_amount),
updated_at_ms = VALUES(updated_at_ms)
`, "test-gift-"+giftID, giftID, coinPrice, giftPointAmount, nowMs, nowMs)
if err != nil {
r.t.Fatalf("seed gift resource failed: %v", err)
}
resourceID, err := result.LastInsertId()
if err != nil {
r.t.Fatalf("read gift resource id failed: %v", err)
}
if _, err := r.schema.DB.ExecContext(context.Background(), `
INSERT INTO gift_configs (
gift_id, resource_id, status, name, sort_order, presentation_json,
gift_type_code, effective_from_ms, effective_to_ms, effect_types_json,
created_by_user_id, updated_by_user_id, created_at_ms, updated_at_ms
) VALUES (?, ?, 'active', ?, 0, JSON_OBJECT(), 'normal', 0, 0, JSON_ARRAY(), 0, 0, ?, ?)
ON DUPLICATE KEY UPDATE
resource_id = VALUES(resource_id),
status = VALUES(status),
name = VALUES(name),
gift_type_code = VALUES(gift_type_code),
effective_from_ms = VALUES(effective_from_ms),
effective_to_ms = VALUES(effective_to_ms),
updated_at_ms = VALUES(updated_at_ms)
`, giftID, resourceID, giftID, nowMs, nowMs); err != nil {
r.t.Fatalf("seed gift config failed: %v", err)
}
if _, err := r.schema.DB.ExecContext(context.Background(), `
INSERT INTO gift_config_regions (gift_id, region_id, created_at_ms, updated_at_ms)
VALUES (?, 0, ?, ?)
ON DUPLICATE KEY UPDATE updated_at_ms = VALUES(updated_at_ms)
`, giftID, nowMs, nowMs); err != nil {
r.t.Fatalf("seed gift config region failed: %v", err)
}
_, err = r.schema.DB.ExecContext(context.Background(), `
INSERT INTO wallet_gift_prices (
gift_id, price_version, status, coin_price, gift_point_amount, heat_value,
effective_at_ms, created_at_ms, updated_at_ms
) VALUES (?, ?, 'active', ?, ?, ?, 0, ?, ?)
ON DUPLICATE KEY UPDATE
status = VALUES(status),
coin_price = VALUES(coin_price),
gift_point_amount = VALUES(gift_point_amount),
heat_value = VALUES(heat_value),
updated_at_ms = VALUES(updated_at_ms)
`, giftID, priceVersion, coinPrice, giftPointAmount, heatValue, nowMs, nowMs)
if err != nil {
r.t.Fatalf("seed wallet gift price failed: %v", err)
}
}
// SetGiftType updates a seeded gift config to the requested gift type.
func (r *Repository) SetGiftType(giftID string, giftTypeCode string) {
r.t.Helper()
_, err := r.schema.DB.ExecContext(context.Background(), `
UPDATE gift_configs
SET gift_type_code = ?, updated_at_ms = ?
WHERE gift_id = ?`, giftTypeCode, time.Now().UnixMilli(), giftID)
if err != nil {
r.t.Fatalf("set gift type failed: %v", err)
}
}
// SetGiftDiamondRatio configures the host-period diamond ratio for a sender region and gift type.
func (r *Repository) SetGiftDiamondRatio(regionID int64, giftTypeCode string, ratioPercent string) {
r.t.Helper()
nowMs := time.Now().UnixMilli()
_, err := r.schema.DB.ExecContext(context.Background(), `
INSERT INTO gift_diamond_ratio_configs (
region_id, gift_type_code, status, ratio_percent,
created_by_admin_id, updated_by_admin_id, created_at_ms, updated_at_ms
) VALUES (?, ?, 'active', ?, 0, 0, ?, ?)
ON DUPLICATE KEY UPDATE
status = VALUES(status),
ratio_percent = VALUES(ratio_percent),
updated_at_ms = VALUES(updated_at_ms)`,
regionID, giftTypeCode, ratioPercent, nowMs, nowMs)
if err != nil {
r.t.Fatalf("set gift diamond ratio failed: %v", err)
}
}
// SetRechargePolicy 配置区域充值汇率,用于币商转账充值口径测试。
func (r *Repository) SetRechargePolicy(regionID int64, policyVersion string, coinAmount int64, usdMinorAmount int64) {
r.t.Helper()
nowMs := time.Now().UnixMilli()
_, err := r.schema.DB.ExecContext(context.Background(), `
INSERT INTO wallet_recharge_policies (
region_id, policy_version, status, currency_code, coin_amount, usd_minor_amount,
effective_from_ms, created_at_ms, updated_at_ms
) VALUES (?, ?, 'active', 'USD', ?, ?, 0, ?, ?)
ON DUPLICATE KEY UPDATE
status = VALUES(status),
currency_code = VALUES(currency_code),
coin_amount = VALUES(coin_amount),
usd_minor_amount = VALUES(usd_minor_amount),
updated_at_ms = VALUES(updated_at_ms)
`, regionID, policyVersion, coinAmount, usdMinorAmount, nowMs, nowMs)
if err != nil {
r.t.Fatalf("seed wallet recharge policy failed: %v", err)
}
}
// SetRechargeStats prepares cumulative recharge stats for VIP gate tests.
func (r *Repository) SetRechargeStats(userID int64, totalCoinAmount int64) {
r.t.Helper()
nowMs := time.Now().UnixMilli()
_, err := r.schema.DB.ExecContext(context.Background(), `
INSERT INTO wallet_user_recharge_stats (
user_id, recharge_count, first_transaction_id, first_recharged_at_ms,
total_coin_amount, total_usd_minor_amount, created_at_ms, updated_at_ms
) VALUES (?, 1, 'test_recharge', ?, ?, 0, ?, ?)
ON DUPLICATE KEY UPDATE
total_coin_amount = VALUES(total_coin_amount),
updated_at_ms = VALUES(updated_at_ms)
`, userID, nowMs, totalCoinAmount, nowMs, nowMs)
if err != nil {
r.t.Fatalf("seed wallet recharge stats failed: %v", err)
}
}
// SetHostSalaryPolicy seeds the wallet runtime policy snapshot used by salary settlement tests.
func (r *Repository) SetHostSalaryPolicy(policy ledger.HostSalaryPolicy) {
r.t.Helper()
nowMs := time.Now().UnixMilli()
if policy.PolicyID == 0 {
policy.PolicyID = 1
}
if policy.Status == "" {
policy.Status = "active"
}
if policy.SettlementMode == "" {
policy.SettlementMode = ledger.HostSalarySettlementModeDaily
}
if policy.SettlementTriggerMode == "" {
// 测试默认覆盖生产 cron 路径;需要验证人工政策时再显式传 manual。
policy.SettlementTriggerMode = ledger.HostSalarySettlementTriggerAutomatic
}
if policy.GiftCoinToDiamondRatio == "" {
policy.GiftCoinToDiamondRatio = "1"
}
if policy.ResidualDiamondToUSDRate == "" {
policy.ResidualDiamondToUSDRate = "0"
}
_, err := r.schema.DB.ExecContext(context.Background(), `
INSERT INTO host_agency_salary_policies (
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 (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
ON DUPLICATE KEY UPDATE
name = VALUES(name),
region_id = VALUES(region_id),
status = VALUES(status),
settlement_mode = VALUES(settlement_mode),
settlement_trigger_mode = VALUES(settlement_trigger_mode),
gift_coin_to_diamond_ratio = VALUES(gift_coin_to_diamond_ratio),
residual_diamond_to_usd_rate = VALUES(residual_diamond_to_usd_rate),
effective_from_ms = VALUES(effective_from_ms),
effective_to_ms = VALUES(effective_to_ms),
updated_at_ms = VALUES(updated_at_ms)
`, policy.PolicyID, policy.Name, policy.RegionID, policy.Status, policy.SettlementMode, policy.SettlementTriggerMode, policy.GiftCoinToDiamondRatio, policy.ResidualDiamondToUSDRate, policy.EffectiveFromMs, policy.EffectiveToMs, nowMs, nowMs)
if err != nil {
r.t.Fatalf("seed host salary policy failed: %v", err)
}
if _, err := r.schema.DB.ExecContext(context.Background(), `DELETE FROM host_agency_salary_policy_levels WHERE policy_id = ?`, policy.PolicyID); err != nil {
r.t.Fatalf("clear host salary policy levels failed: %v", err)
}
for _, level := range policy.Levels {
if level.Status == "" {
level.Status = "active"
}
if _, err := r.schema.DB.ExecContext(context.Background(), `
INSERT INTO host_agency_salary_policy_levels (
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 (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`,
policy.PolicyID, level.LevelNo, level.RequiredDiamonds, level.HostSalaryUSDMinor,
level.HostCoinReward, level.AgencySalaryUSDMinor, level.Status, level.SortOrder, nowMs, nowMs,
); err != nil {
r.t.Fatalf("seed host salary policy level %d failed: %v", level.LevelNo, err)
}
}
}
func initDBPath(t testing.TB) string {
t.Helper()
return mysqlschema.InitDBPath(t,
mysqlschema.CallerFile(t, 1),
"..", "..", "..", "deploy", "mysql", "initdb", "001_wallet_service.sql",
)
}
func validateTableName(table string) string {
switch table {
case "wallet_transactions", "wallet_entries", "wallet_outbox", "wallet_accounts", "wallet_recharge_records",
"host_period_diamond_accounts", "host_period_diamond_entries", "host_agency_salary_policies",
"host_agency_salary_policy_levels", "host_salary_settlement_progress", "host_salary_settlement_records",
"coin_seller_stock_records", "gift_diamond_ratio_configs", "wallet_diamond_exchange_rules",
"vip_levels", "user_vip_memberships", "vip_purchase_orders", "user_vip_history",
"resources", "resource_groups", "resource_group_items", "resource_shop_items", "resource_shop_purchase_orders", "gift_type_configs", "gift_configs", "gift_config_regions", "user_gift_wall", "wallet_projection_events",
"resource_grants", "resource_grant_items", "user_resource_entitlements", "user_resource_equipment":
return table
default:
panic(fmt.Sprintf("unsupported wallet test table %q", table))
}
}