hyapp-server/services/activity-service/tools/invite-activity-local-flow/main.go

package main

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

	_ "github.com/go-sql-driver/mysql"
	"google.golang.org/grpc"
	"google.golang.org/grpc/credentials/insecure"
	activityv1 "hyapp.local/api/proto/activity/v1"
	userv1 "hyapp.local/api/proto/user/v1"
	walletv1 "hyapp.local/api/proto/wallet/v1"
	"hyapp/pkg/appcode"
	domain "hyapp/services/activity-service/internal/domain/inviteactivity"
	inviteactivityservice "hyapp/services/activity-service/internal/service/inviteactivity"
	mysqlstorage "hyapp/services/activity-service/internal/storage/mysql"
)

const (
	appCode         = "lalu"
	operatorAdminID = int64(900100100)
	tierStatus      = "active"
)

type walletEntrySummary struct {
	Count int64 `json:"count"`
	Coins int64 `json:"coins"`
}

func main() {
	var (
		activityAddr = flag.String("activity_addr", "127.0.0.1:14006", "activity-service gRPC address")
		walletAddr   = flag.String("wallet_addr", "127.0.0.1:13004", "wallet-service gRPC address")
		activityDSN  = flag.String("activity_dsn", "hyapp:hyapp@tcp(127.0.0.1:23306)/hyapp_activity?parseTime=true&charset=utf8mb4&loc=UTC", "activity MySQL DSN")
		walletDSN    = flag.String("wallet_dsn", "hyapp:hyapp@tcp(127.0.0.1:23306)/hyapp_wallet?parseTime=true&charset=utf8mb4&loc=UTC", "wallet MySQL DSN")
	)
	flag.Parse()

	ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
	defer cancel()
	ctx = appcode.WithContext(ctx, appCode)

	conn, err := grpc.DialContext(ctx, *activityAddr, grpc.WithTransportCredentials(insecure.NewCredentials()))
	must("dial activity-service", err)
	defer conn.Close()
	adminClient := activityv1.NewAdminInviteActivityRewardServiceClient(conn)
	rewardClient := activityv1.NewInviteActivityRewardServiceClient(conn)
	walletConn, err := grpc.DialContext(ctx, *walletAddr, grpc.WithTransportCredentials(insecure.NewCredentials()))
	must("dial wallet-service", err)
	defer walletConn.Close()
	walletClient := walletv1.NewWalletServiceClient(walletConn)

	activityRepo, err := mysqlstorage.Open(ctx, *activityDSN)
	must("open activity mysql", err)
	defer activityRepo.Close()
	must("migrate activity mysql", activityRepo.Migrate(ctx))
	walletDB, err := sql.Open("mysql", *walletDSN)
	must("open wallet mysql", err)
	defer walletDB.Close()
	must("ping wallet mysql", walletDB.PingContext(ctx))

	runID := fmt.Sprintf("iar_%d", time.Now().UnixMilli())
	inviterUserID := int64(880000000000) + time.Now().Unix()%10000000
	invitedUserID := inviterUserID + 9001
	now := time.Now().UTC()
	cycle := inviteactivityservice.CycleForTime(now)

	config := updateConfig(ctx, adminClient, runID)
	rechargeTier := findTier(config.GetTiers(), domain.RewardTypeRecharge, runID+"_recharge_160k")
	validTier := findTier(config.GetTiers(), domain.RewardTypeValidInvite, runID+"_valid_1")

	service := inviteactivityservice.New(activityRepo, walletClient, fakeInviteAttributionSource{inviterUserID: inviterUserID})
	consumeInvite(ctx, service, runID, inviterUserID, invitedUserID, now)
	afterInviteInviterStatus := getStatus(ctx, rewardClient, inviterUserID)
	afterInviteInviteeStatus := getStatus(ctx, rewardClient, invitedUserID)
	inviterInviteClaim := findClaim(afterInviteInviterStatus.GetClaims(), domain.RewardTypeInviteInviter)
	inviteeInviteClaim := findClaim(afterInviteInviteeStatus.GetClaims(), domain.RewardTypeInvitee)
	assert(inviterInviteClaim.GetRewardCoinAmount() == 700, "inviter per invite reward should be 700 coins")
	assert(inviteeInviteClaim.GetRewardCoinAmount() == 200, "invitee per invite reward should be 200 coins")
	consumeRecharge(ctx, service, runID, inviterUserID, 1, "google_play", 80000, now.Add(time.Minute))
	consumeRecharge(ctx, service, runID, inviterUserID, 2, "mifapay", 40000, now.Add(2*time.Minute))
	consumeRecharge(ctx, service, runID, inviterUserID, 3, "coin_seller_transfer", 40000, now.Add(3*time.Minute))
	consumeValidInvite(ctx, service, runID, inviterUserID, 1, now.Add(4*time.Minute))

	beforeClaim := getStatus(ctx, rewardClient, inviterUserID)
	assert(beforeClaim.GetProgress().GetTotalRechargeCoinAmount() == 160000, "invite recharge progress should be 160000 coins")
	assert(beforeClaim.GetProgress().GetValidInviteCount() == 1, "valid invite progress should be 1")
	leaderboard := listLeaderboard(ctx, rewardClient)
	assert(len(leaderboard.GetEntries()) > 0, "leaderboard should include at least one entry")
	leaderboardEntry := findLeaderboardEntry(leaderboard.GetEntries(), inviterUserID)
	assert(leaderboardEntry != nil, "leaderboard should include inviter")
	assert(leaderboardEntry.GetValidInviteCount() == 1, "leaderboard valid invite count should be 1")

	rechargeClaim := claim(ctx, rewardClient, inviterUserID, domain.RewardTypeRecharge, rechargeTier.GetTierId(), runID+"_claim_recharge")
	validClaim := claim(ctx, rewardClient, inviterUserID, domain.RewardTypeValidInvite, validTier.GetTierId(), runID+"_claim_valid")
	afterClaim := getStatus(ctx, rewardClient, inviterUserID)
	assert(afterClaim.GetProgress().GetClaimedRewardCoinAmount() == 3500, "claimed reward progress should be 3500 coins")

	walletInviteInviter := queryWalletEntry(ctx, walletDB, inviterInviteClaim.GetWalletTransactionId())
	walletInvitee := queryWalletEntry(ctx, walletDB, inviteeInviteClaim.GetWalletTransactionId())
	walletRecharge := queryWalletEntry(ctx, walletDB, rechargeClaim.GetWalletTransactionId())
	walletValid := queryWalletEntry(ctx, walletDB, validClaim.GetWalletTransactionId())
	assert(walletInviteInviter.Coins == 700, "wallet inviter per invite reward should credit 700 coins")
	assert(walletInvitee.Coins == 200, "wallet invitee per invite reward should credit 200 coins")
	assert(walletRecharge.Coins == 2500, "wallet recharge tier reward should credit 2500 coins")
	assert(walletValid.Coins == 300, "wallet valid invite reward should credit 300 coins")

	out := map[string]any{
		"run_id":          runID,
		"inviter_user_id": inviterUserID,
		"invited_user_id": invitedUserID,
		"cycle":           cycle.Key,
		"progress_before_claim": map[string]int64{
			"total_recharge_coin_amount": beforeClaim.GetProgress().GetTotalRechargeCoinAmount(),
			"valid_invite_count":         beforeClaim.GetProgress().GetValidInviteCount(),
		},
		"leaderboard_entry": map[string]any{
			"rank_no":                    leaderboardEntry.GetRankNo(),
			"user_id":                    leaderboardEntry.GetUserId(),
			"valid_invite_count":         leaderboardEntry.GetValidInviteCount(),
			"total_recharge_coin_amount": leaderboardEntry.GetTotalRechargeCoinAmount(),
		},
		"claims": []map[string]any{
			{"reward_type": inviterInviteClaim.GetRewardType(), "tier_code": inviterInviteClaim.GetTierCode(), "reward_coin_amount": inviterInviteClaim.GetRewardCoinAmount(), "wallet_transaction_id": inviterInviteClaim.GetWalletTransactionId()},
			{"reward_type": inviteeInviteClaim.GetRewardType(), "tier_code": inviteeInviteClaim.GetTierCode(), "reward_coin_amount": inviteeInviteClaim.GetRewardCoinAmount(), "wallet_transaction_id": inviteeInviteClaim.GetWalletTransactionId()},
			{"reward_type": rechargeClaim.GetRewardType(), "tier_code": rechargeClaim.GetTierCode(), "reward_coin_amount": rechargeClaim.GetRewardCoinAmount(), "wallet_transaction_id": rechargeClaim.GetWalletTransactionId()},
			{"reward_type": validClaim.GetRewardType(), "tier_code": validClaim.GetTierCode(), "reward_coin_amount": validClaim.GetRewardCoinAmount(), "wallet_transaction_id": validClaim.GetWalletTransactionId()},
		},
		"wallet_entries": map[string]walletEntrySummary{
			"invite_inviter": walletInviteInviter,
			"invite_invitee": walletInvitee,
			"recharge":       walletRecharge,
			"valid_invite":   walletValid,
		},
		"progress_after_claim": map[string]int64{
			"claimed_reward_coin_amount": afterClaim.GetProgress().GetClaimedRewardCoinAmount(),
		},
	}
	encoded, _ := json.MarshalIndent(out, "", "  ")
	fmt.Println(string(encoded))
}

func updateConfig(ctx context.Context, client activityv1.AdminInviteActivityRewardServiceClient, runID string) *activityv1.InviteActivityRewardConfig {
	resp, err := client.UpdateInviteActivityRewardConfig(ctx, &activityv1.UpdateInviteActivityRewardConfigRequest{
		Meta:                             meta(runID + "_update_config"),
		Enabled:                          true,
		PerInviteInviterRewardCoinAmount: 700,
		PerInviteInviteeRewardCoinAmount: 200,
		OperatorAdminId:                  operatorAdminID,
		Tiers: []*activityv1.InviteActivityRewardTier{
			{RewardType: domain.RewardTypeRecharge, TierCode: runID + "_recharge_80k", TierName: "Local recharge 80k", ThresholdCoinAmount: 80000, RewardCoinAmount: 1000, Status: tierStatus, SortOrder: 1},
			{RewardType: domain.RewardTypeRecharge, TierCode: runID + "_recharge_160k", TierName: "Local recharge 160k", ThresholdCoinAmount: 160000, RewardCoinAmount: 2500, Status: tierStatus, SortOrder: 2},
			{RewardType: domain.RewardTypeValidInvite, TierCode: runID + "_valid_1", TierName: "Local valid invite 1", ThresholdValidInviteCount: 1, RewardCoinAmount: 300, Status: tierStatus, SortOrder: 1},
			{RewardType: domain.RewardTypeValidInvite, TierCode: runID + "_valid_2", TierName: "Local valid invite 2", ThresholdValidInviteCount: 2, RewardCoinAmount: 900, Status: tierStatus, SortOrder: 2},
		},
	})
	must("update invite activity config", err)
	return resp.GetConfig()
}

func consumeInvite(ctx context.Context, service *inviteactivityservice.Service, runID string, inviterUserID int64, invitedUserID int64, occurredAt time.Time) {
	err := service.ConsumeInviteEvent(ctx, domain.InviteEvent{
		AppCode:       appCode,
		EventID:       runID + "_invite_bound_1",
		EventType:     domain.EventTypeUserInvited,
		InvitedUserID: invitedUserID,
		InviterUserID: inviterUserID,
		InviteCode:    runID,
		Source:        "invite_h5",
		OccurredAtMS:  occurredAt.UnixMilli(),
	})
	must("consume invite bound event", err)
}

func consumeRecharge(ctx context.Context, service *inviteactivityservice.Service, runID string, inviterUserID int64, index int64, rechargeType string, coins int64, occurredAt time.Time) {
	err := service.ConsumeWalletRechargeEvent(ctx, domain.RechargeEvent{
		AppCode:            appCode,
		EventID:            fmt.Sprintf("%s_recharge_%d", runID, index),
		EventType:          domain.EventTypeWalletRechargeRecorded,
		TransactionID:      fmt.Sprintf("%s_tx_%d", runID, index),
		CommandID:          fmt.Sprintf("%s_cmd_%d", runID, index),
		InvitedUserID:      inviterUserID + 100 + index,
		InviterUserID:      inviterUserID,
		RechargeCoinAmount: coins,
		RechargeType:       rechargeType,
		OccurredAtMS:       occurredAt.UnixMilli(),
	})
	must("consume invite recharge "+rechargeType, err)
}

func consumeValidInvite(ctx context.Context, service *inviteactivityservice.Service, runID string, inviterUserID int64, index int64, occurredAt time.Time) {
	err := service.ConsumeValidInviteEvent(ctx, domain.ValidInviteEvent{
		AppCode:       appCode,
		EventID:       fmt.Sprintf("%s_valid_%d", runID, index),
		EventType:     domain.EventTypeUserInviteBecameValid,
		InvitedUserID: inviterUserID + 200 + index,
		InviterUserID: inviterUserID,
		OccurredAtMS:  occurredAt.UnixMilli(),
	})
	must("consume valid invite", err)
}

func getStatus(ctx context.Context, client activityv1.InviteActivityRewardServiceClient, userID int64) *activityv1.InviteActivityRewardStatus {
	resp, err := client.GetInviteActivityRewardStatus(ctx, &activityv1.GetInviteActivityRewardStatusRequest{
		Meta:   meta("get_status"),
		UserId: userID,
	})
	must("get invite activity status", err)
	return resp.GetStatus()
}

func listLeaderboard(ctx context.Context, client activityv1.InviteActivityRewardServiceClient) *activityv1.ListInviteActivityLeaderboardResponse {
	resp, err := client.ListInviteActivityLeaderboard(ctx, &activityv1.ListInviteActivityLeaderboardRequest{
		Meta:     meta("list_leaderboard"),
		Page:     1,
		PageSize: 50,
	})
	must("list invite activity leaderboard", err)
	return resp
}

func claim(ctx context.Context, client activityv1.InviteActivityRewardServiceClient, userID int64, rewardType string, tierID int64, commandID string) *activityv1.InviteActivityRewardClaim {
	resp, err := client.ClaimInviteActivityReward(ctx, &activityv1.ClaimInviteActivityRewardRequest{
		Meta:       meta(commandID),
		UserId:     userID,
		RewardType: rewardType,
		TierId:     tierID,
		CommandId:  commandID,
	})
	must("claim invite activity reward "+rewardType, err)
	return resp.GetClaim()
}

func queryWalletEntry(ctx context.Context, db *sql.DB, transactionID string) walletEntrySummary {
	var summary walletEntrySummary
	err := db.QueryRowContext(ctx, `
		SELECT COUNT(*), COALESCE(SUM(available_delta), 0)
		FROM wallet_entries
		WHERE app_code = ? AND transaction_id = ? AND asset_type = 'COIN'`,
		appCode, transactionID,
	).Scan(&summary.Count, &summary.Coins)
	must("query wallet entry "+transactionID, err)
	return summary
}

func findTier(tiers []*activityv1.InviteActivityRewardTier, rewardType string, tierCode string) *activityv1.InviteActivityRewardTier {
	for _, tier := range tiers {
		if tier.GetRewardType() == rewardType && tier.GetTierCode() == tierCode {
			return tier
		}
	}
	log.Fatalf("tier not found: %s %s", rewardType, tierCode)
	return nil
}

func findClaim(claims []*activityv1.InviteActivityRewardClaim, rewardType string) *activityv1.InviteActivityRewardClaim {
	for _, claim := range claims {
		if claim.GetRewardType() == rewardType {
			return claim
		}
	}
	log.Fatalf("claim not found: %s", rewardType)
	return nil
}

func findLeaderboardEntry(entries []*activityv1.InviteActivityLeaderboardEntry, userID int64) *activityv1.InviteActivityLeaderboardEntry {
	for _, entry := range entries {
		if entry.GetUserId() == userID {
			return entry
		}
	}
	return nil
}

func meta(requestID string) *activityv1.RequestMeta {
	return &activityv1.RequestMeta{
		RequestId: requestID,
		Caller:    "invite-activity-local-flow",
		AppCode:   appCode,
		SentAtMs:  time.Now().UTC().UnixMilli(),
	}
}

func assert(ok bool, message string) {
	if !ok {
		log.Fatal(message)
	}
}

func must(step string, err error) {
	if err != nil {
		log.Fatalf("%s: %v", step, err)
	}
}

type fakeInviteAttributionSource struct {
	inviterUserID int64
}

func (s fakeInviteAttributionSource) GetInviteAttribution(_ context.Context, invitedUserID int64) (*userv1.InviteAttribution, error) {
	return &userv1.InviteAttribution{
		Found:         true,
		AppCode:       appCode,
		InvitedUserId: invitedUserID,
		InviterUserId: s.inviterUserID,
	}, nil
}