dashboard-cdc-worker/internal/dashboard/mapper.go

package dashboard

import (
	"context"
	"strings"
	"time"

	"github.com/shopspring/decimal"
)

type Mapper struct {
	store                   Store
	sysOrigins              map[string]struct{}
	unknownCountry          string
	userCreateDisplayOffset time.Duration
}

func NewMapper(store Store, sysOrigins []string, unknownCountry string,
	userCreateDisplayOffset time.Duration) *Mapper {
	origins := make(map[string]struct{}, len(sysOrigins))
	for _, origin := range sysOrigins {
		origins[strings.TrimSpace(origin)] = struct{}{}
	}
	return &Mapper{
		store:                   store,
		sysOrigins:              origins,
		unknownCountry:          unknownCountry,
		userCreateDisplayOffset: userCreateDisplayOffset,
	}
}

func (m *Mapper) Map(ctx context.Context, schema string, table string, action string, row Row) ([]Contribution, error) {
	switch table {
	case "user_base_info":
		return m.mapUser(ctx, schema, table, action, row)
	case "order_purchase_history":
		return m.mapOfficialRecharge(ctx, schema, table, action, row)
	case "order_user_purchase_pay":
		return m.mapThirdPartyRecharge(ctx, schema, table, action, row)
	case "game_lucky_gift_count":
		return m.mapLuckyGift(ctx, schema, table, action, row)
	case "game_lucky_box_count":
		return m.mapLuckyBox(ctx, schema, table, action, row)
	case "user_freight_balance_running_water":
		return m.mapDealerRecharge(ctx, schema, table, action, row)
	case "user_salary_account_running_water", "user_salary_diamond_running_water":
		return m.mapSalary(ctx, schema, table, action, row)
	default:
		if strings.HasPrefix(table, "wallet_gold_asset_record_") {
			return m.mapWalletGame(ctx, schema, table, action, row)
		}
		return nil, nil
	}
}

func (m *Mapper) mapUser(_ context.Context, schema string, table string, action string, row Row) ([]Contribution, error) {
	user := UserCountryFromRow(row, m.unknownCountry)
	if !m.allowed(user.SysOrigin) || user.IsDeleted || user.UserID == 0 || user.CreatedAt.IsZero() {
		return nil, nil
	}
	displayCreateTime := user.CreatedAt.Add(m.userCreateDisplayOffset)
	return []Contribution{{
		SourceSchema:   schema,
		SourceTable:    table,
		SourcePK:       rowPK(row),
		SourceAction:   action,
		DeltaSign:      signForAction(action),
		SysOrigin:      user.SysOrigin,
		UserID:         user.UserID,
		Country:        user.Country,
		EventTime:      displayCreateTime,
		CountryNewUser: 1,
	}}, nil
}

func (m *Mapper) mapOfficialRecharge(ctx context.Context, schema string, table string, action string, row Row) ([]Contribution, error) {
	if row.String("evn") != "PROD" || row.Bool("is_trial_period") || row.String("status") != "COMPLETE" {
		return nil, nil
	}
	sysOrigin := row.String("sys_origin")
	if !m.allowed(sysOrigin) {
		return nil, nil
	}
	user, ok, err := m.user(ctx, row.Int64("user_id"))
	if err != nil || !ok || user.IsDeleted {
		return nil, err
	}
	if user.SysOrigin != sysOrigin {
		return nil, nil
	}
	amount := row.Decimal("unit_price")
	payPlatform := row.String("pay_platform")
	c := m.base(schema, table, action, row, user, row.Time("create_time"))
	if payPlatform != "MIFA_PAY" {
		c.OfficialRecharge = amount
	}
	if payPlatform == "GOOGLE" {
		c.GoogleRecharge = amount
		c.UserRecharge = amount
	}
	c.NewUserRechargeCandidate = amount
	c.RechargeDetail = rechargeDetail("OFFICIAL", rowPK(row), payPlatform, row, user, amount,
		row.Decimal("coin_quantity"), "")
	return []Contribution{c}, nil
}

func (m *Mapper) mapThirdPartyRecharge(ctx context.Context, schema string, table string, action string, row Row) ([]Contribution, error) {
	if row.String("evn") != "PROD" ||
		row.String("pay_status") != "SUCCESSFUL" || row.String("receipt_type") != "PAYMENT" ||
		row.String("refund_status") != "NONE" {
		return nil, nil
	}
	items, err := m.amountContribution(ctx, schema, table, action, row, row.Decimal("compute_usd_amount"), func(c *Contribution, amount decimal.Decimal) {
		c.UserRecharge = amount
		if row.String("factory_code") == "MIFA_PAY" {
			c.MifapayRecharge = amount
		}
	})
	if err != nil || len(items) == 0 {
		return items, err
	}
	sourceType := "THIRD_PARTY"
	if row.String("factory_code") == "MIFA_PAY" {
		sourceType = "MIFAPAY"
	}
	items[0].RechargeDetail = rechargeDetail(sourceType, rowPK(row), row.String("factory_code"), row,
		detailUser(items[0]), row.Decimal("compute_usd_amount"), decimal.Zero, "")
	return items, nil
}

func (m *Mapper) mapDealerRecharge(ctx context.Context, schema string, table string, action string, row Row) ([]Contribution, error) {
	if row.String("origin") != "PURCHASE" || row.Int64("type") != 0 || !row.Decimal("amount").GreaterThan(decimal.Zero) {
		return nil, nil
	}
	items, err := m.amountContribution(ctx, schema, table, action, row, row.Decimal("amount"), func(c *Contribution, amount decimal.Decimal) {
		c.DealerRecharge = amount
		c.NewDealerUserRecharge = amount
	})
	if err != nil || len(items) == 0 {
		return items, err
	}
	items[0].RechargeDetail = rechargeDetail("DEALER", rowPK(row), row.String("recharge_type"), row,
		detailUser(items[0]), row.Decimal("amount"), row.Decimal("quantity"), row.String("remark"))
	return items, nil
}

func (m *Mapper) mapSalary(ctx context.Context, schema string, table string, action string, row Row) ([]Contribution, error) {
	if row.Int64("type") != 1 {
		return nil, nil
	}
	event := row.String("salary_event")
	isExchange := event == "SALARY_EXCHANGE" || event == "SALARY_EXCHANGE_GOLD_COINS" ||
		event == "EXCHANGE_GOLD_COINS" || event == "BILL_EXCHANGE_GOLD_COINS"
	isTransfer := event == "SALARY_TRANSFER"
	if !isExchange && !isTransfer {
		return nil, nil
	}
	return m.amountContribution(ctx, schema, table, action, row, row.Decimal("amount"), func(c *Contribution, amount decimal.Decimal) {
		if isExchange {
			c.SalaryExchange = amount
		}
		if isTransfer {
			c.SalaryTransfer = amount
		}
	})
}

func (m *Mapper) mapLuckyGift(ctx context.Context, schema string, table string, action string, row Row) ([]Contribution, error) {
	return m.amountContribution(ctx, schema, table, action, row, row.Decimal("pay_amount"), func(c *Contribution, amount decimal.Decimal) {
		c.LuckyGiftTotalFlow = amount
		c.LuckyGiftPayout = row.Decimal("award_amount")
		c.LuckyGiftUser = true
	})
}

func (m *Mapper) mapLuckyBox(ctx context.Context, schema string, table string, action string, row Row) ([]Contribution, error) {
	return m.amountContribution(ctx, schema, table, action, row, row.Decimal("pay_amount"), func(c *Contribution, amount decimal.Decimal) {
		payout := row.Decimal("gift_amount")
		c.GameTotalFlow = amount
		c.GamePayout = payout
		c.GameUser = true
		c.GameProvider = "INTERNAL"
		c.GameID = "LUCKY_BOX"
		c.GameName = "Lucky Box"
		c.GameConsume = amount
		c.GamePayoutBy = payout
		c.GameProfit = amount.Sub(payout)
		c.GameOrder = 1
	})
}