hyapp-server/tools/cumulative-recharge-local-flow/main.go

package main

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

	_ "github.com/go-sql-driver/mysql"
	"google.golang.org/grpc"
	"google.golang.org/grpc/credentials/insecure"
	activityv1 "hyapp.local/api/proto/activity/v1"
	walletv1 "hyapp.local/api/proto/wallet/v1"
)

const (
	appCode              = "lalu"
	operatorAdminID      = int64(900100100)
	rechargeSource       = "cumulative_recharge_reward"
	groupItemWalletAsset = "wallet_asset"
	assetCoin            = "COIN"
	statusActive         = "active"
	statusDisabled       = "disabled"
	tierStatusInactive   = "inactive"
)

type walletOutboxFact struct {
	EventID        string
	TransactionID  string
	CommandID      string
	PayloadJSON    string
	CreatedAtMS    int64
	AvailableDelta int64
}

type cycle struct {
	Key     string
	StartMS int64
	EndMS   int64
}

func main() {
	var (
		walletAddr   = flag.String("wallet_addr", "127.0.0.1:13004", "wallet-service gRPC address")
		activityAddr = flag.String("activity_addr", "127.0.0.1:13006", "activity-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")
		mode         = flag.String("mode", "happy", "flow mode: happy or boundary")
	)
	flag.Parse()

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

	walletConn, err := grpc.DialContext(ctx, *walletAddr, grpc.WithTransportCredentials(insecure.NewCredentials()))
	must("dial wallet-service", err)
	defer walletConn.Close()
	activityConn, err := grpc.DialContext(ctx, *activityAddr, grpc.WithTransportCredentials(insecure.NewCredentials()))
	must("dial activity-service", err)
	defer activityConn.Close()

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

	walletClient := walletv1.NewWalletServiceClient(walletConn)
	rewardClient := activityv1.NewCumulativeRechargeRewardServiceClient(activityConn)
	adminClient := activityv1.NewAdminCumulativeRechargeRewardServiceClient(activityConn)

	if *mode == "boundary" {
		runBoundaryFlow(ctx, walletClient, rewardClient, adminClient, activityDB, walletDB)
		return
	}

	runID := fmt.Sprintf("crr_%d", time.Now().UnixMilli())
	targetUserID := int64(760000000000) + time.Now().Unix()%10000000
	sellerUserID := targetUserID + 100000000
	now := time.Now().UTC()
	currentCycle := cycleForTime(now)
	nextCycle := cycleForTime(time.UnixMilli(currentCycle.EndMS).UTC().Add(2 * time.Hour))

	group100 := createCoinRewardGroup(ctx, walletClient, runID, "100", 1000)
	group200 := createCoinRewardGroup(ctx, walletClient, runID, "200", 2000)
	updateConfig(ctx, adminClient, runID, group100, group200)

	google := consumeRecharge(ctx, rewardClient, runID, "google", targetUserID, "google_play", 5_400_000, 6_000, now.Add(1*time.Minute))
	assert(len(google.Grants) == 0 && google.GetReason() == "no_tier_matched", "google $60 should not grant a tier")
	statusAfterGoogle := getStatus(ctx, rewardClient, targetUserID)
	assert(statusAfterGoogle.GetProgress().GetTotalUsdMinor() == 6_000, "google progress should be $60.00")

	mifapay := consumeRecharge(ctx, rewardClient, runID, "mifapay", targetUserID, "mifapay", 4_500_000, 5_000, now.Add(2*time.Minute))
	assertGrantThresholds("mifapay should unlock only $100 tier", mifapay.GetGrants(), []int64{10_000})
	statusAfterMifapay := getStatus(ctx, rewardClient, targetUserID)
	assert(statusAfterMifapay.GetProgress().GetTotalUsdMinor() == 11_000, "google+mifapay progress should be $110.00")

	transfer := runCoinSellerTransfer(ctx, walletClient, walletDB, runID, sellerUserID, targetUserID)
	coinSellerUSDMinor := transfer.AvailableDelta * 100 / 90_000
	assert(transfer.AvailableDelta == 9_000_000 && coinSellerUSDMinor == 10_000, "coin seller transfer should convert 9,000,000 coins to $100.00")
	coinSeller := consumeRechargeFromWalletFact(ctx, rewardClient, targetUserID, transfer, coinSellerUSDMinor)
	assertGrantThresholds("coin seller transfer should unlock only $200 tier", coinSeller.GetGrants(), []int64{20_000})
	statusAfterCoinSeller := getStatus(ctx, rewardClient, targetUserID)
	assert(statusAfterCoinSeller.GetProgress().GetTotalUsdMinor() == 21_000, "current cycle progress should be $210.00")

	nextWeek := consumeRecharge(ctx, rewardClient, runID, "nextweek-google", targetUserID, "google_play", 9_000_000, 10_000, time.UnixMilli(nextCycle.StartMS).UTC().Add(time.Hour))
	assertGrantThresholds("next week $100 should unlock only new $100 tier", nextWeek.GetGrants(), []int64{10_000})

	currentProgress := queryProgress(ctx, activityDB, targetUserID, currentCycle.Key)
	nextProgress := queryProgress(ctx, activityDB, targetUserID, nextCycle.Key)
	assert(currentProgress == 21_000, "current cycle DB progress should stay $210.00")
	assert(nextProgress == 10_000, "next cycle DB progress should reset to $100.00")

	currentGrants := queryGrantSummary(ctx, activityDB, targetUserID, currentCycle.Key)
	nextGrants := queryGrantSummary(ctx, activityDB, targetUserID, nextCycle.Key)
	assert(currentGrants.Count == 2 && currentGrants.Granted == 2, "current cycle should have two granted tiers")
	assert(nextGrants.Count == 1 && nextGrants.Granted == 1, "next cycle should have one granted tier")

	walletReward := queryWalletRewardSummary(ctx, walletDB, targetUserID)
	assert(walletReward.Count == 3 && walletReward.CoinDelta == 4_000, "wallet should receive three reward group grants totaling 4,000 COIN")

	out := map[string]any{
		"run_id":         runID,
		"target_user_id": targetUserID,
		"seller_user_id": sellerUserID,
		"current_cycle":  currentCycle.Key,
		"next_cycle":     nextCycle.Key,
		"events": []map[string]any{
			{"type": "google_play", "usd_minor": 6000, "grant_thresholds": thresholds(google.GetGrants())},
			{"type": "mifapay", "usd_minor": 5000, "grant_thresholds": thresholds(mifapay.GetGrants())},
			{"type": "coin_seller_transfer", "coins": transfer.AvailableDelta, "usd_minor": coinSellerUSDMinor, "grant_thresholds": thresholds(coinSeller.GetGrants())},
			{"type": "next_week_google_play", "usd_minor": 10000, "grant_thresholds": thresholds(nextWeek.GetGrants())},
		},
		"progress_usd_minor": map[string]int64{currentCycle.Key: currentProgress, nextCycle.Key: nextProgress},
		"activity_grants":    map[string]grantSummary{currentCycle.Key: currentGrants, nextCycle.Key: nextGrants},
		"wallet_reward":      walletReward,
	}
	encoded, _ := json.MarshalIndent(out, "", "  ")
	fmt.Println(string(encoded))
}

func createCoinRewardGroup(ctx context.Context, client walletv1.WalletServiceClient, runID string, suffix string, coinAmount int64) int64 {
	resp, err := client.CreateResourceGroup(ctx, &walletv1.CreateResourceGroupRequest{
		RequestId:      fmt.Sprintf("%s_create_group_%s", runID, suffix),
		AppCode:        appCode,
		GroupCode:      fmt.Sprintf("%s_group_%s", runID, suffix),
		Name:           fmt.Sprintf("CRR local reward %s", suffix),
		Status:         statusActive,
		Description:    "local cumulative recharge flow verification",
		SortOrder:      1,
		OperatorUserId: operatorAdminID,
		Items: []*walletv1.ResourceGroupItemInput{{
			ItemType:          groupItemWalletAsset,
			WalletAssetType:   assetCoin,
			WalletAssetAmount: coinAmount,
			SortOrder:         1,
		}},
	})
	must("create wallet reward group "+suffix, err)
	assert(resp.GetGroup().GetGroupId() > 0, "created resource group id should be positive")
	return resp.GetGroup().GetGroupId()
}

func updateConfig(ctx context.Context, client activityv1.AdminCumulativeRechargeRewardServiceClient, runID string, group100 int64, group200 int64) {
	_, err := updateConfigWithTiers(ctx, client, runID, true, []*activityv1.CumulativeRechargeRewardTier{
		{TierCode: runID + "_100", TierName: "Local $100", ThresholdUsdMinor: 10_000, ResourceGroupId: group100, Status: statusActive, SortOrder: 1},
		{TierCode: runID + "_200", TierName: "Local $200", ThresholdUsdMinor: 20_000, ResourceGroupId: group200, Status: statusActive, SortOrder: 2},
	})
	must("update cumulative recharge config", err)
}

func updateConfigWithTiers(ctx context.Context, client activityv1.AdminCumulativeRechargeRewardServiceClient, runID string, enabled bool, tiers []*activityv1.CumulativeRechargeRewardTier) (*activityv1.UpdateCumulativeRechargeRewardConfigResponse, error) {
	return client.UpdateCumulativeRechargeRewardConfig(ctx, &activityv1.UpdateCumulativeRechargeRewardConfigRequest{
		Meta:            meta(runID + "_update_config"),
		Enabled:         enabled,
		OperatorAdminId: operatorAdminID,
		Tiers:           tiers,
	})
}

func consumeRecharge(ctx context.Context, client activityv1.CumulativeRechargeRewardServiceClient, runID string, name string, userID int64, rechargeType string, coinAmount int64, usdMinor int64, occurredAt time.Time) *activityv1.ConsumeCumulativeRechargeRewardResponse {
	resp, err := consumeRechargeAllowError(ctx, client, runID, name, userID, rechargeType, coinAmount, usdMinor, occurredAt)
	must("consume recharge "+name, err)
	return resp
}

func consumeRechargeAllowError(ctx context.Context, client activityv1.CumulativeRechargeRewardServiceClient, runID string, name string, userID int64, rechargeType string, coinAmount int64, usdMinor int64, occurredAt time.Time) (*activityv1.ConsumeCumulativeRechargeRewardResponse, error) {
	payload, _ := json.Marshal(map[string]any{
		"recharge_type":      rechargeType,
		"coin_amount":        coinAmount,
		"recharge_usd_minor": usdMinor,
	})
	resp, err := client.ConsumeCumulativeRechargeReward(ctx, &activityv1.ConsumeCumulativeRechargeRewardRequest{
		Meta:               meta(runID + "_" + name),
		EventId:            fmt.Sprintf("%s_event_%s", runID, name),
		TransactionId:      fmt.Sprintf("%s_tx_%s", runID, name),
		CommandId:          fmt.Sprintf("%s_cmd_%s", runID, name),
		UserId:             userID,
		RechargeCoinAmount: coinAmount,
		RechargeSequence:   time.Now().UnixNano(),
		RechargeType:       rechargeType,
		QualifyingUsdMinor: usdMinor,
		PayloadJson:        string(payload),
		OccurredAtMs:       occurredAt.UnixMilli(),
	})
	return resp, err
}

func consumeRechargeFromWalletFact(ctx context.Context, client activityv1.CumulativeRechargeRewardServiceClient, userID int64, fact walletOutboxFact, usdMinor int64) *activityv1.ConsumeCumulativeRechargeRewardResponse {
	resp, err := client.ConsumeCumulativeRechargeReward(ctx, &activityv1.ConsumeCumulativeRechargeRewardRequest{
		Meta:               meta("coin_seller_wallet_fact"),
		EventId:            fact.EventID,
		TransactionId:      fact.TransactionID,
		CommandId:          fact.CommandID,
		UserId:             userID,
		RechargeCoinAmount: fact.AvailableDelta,
		RechargeType:       "coin_seller_transfer",
		QualifyingUsdMinor: usdMinor,
		PayloadJson:        fact.PayloadJSON,
		OccurredAtMs:       fact.CreatedAtMS,
	})
	must("consume coin seller wallet fact", err)
	return resp
}

func runCoinSellerTransfer(ctx context.Context, client walletv1.WalletServiceClient, db *sql.DB, runID string, sellerUserID int64, targetUserID int64) walletOutboxFact {
	_, err := client.AdminCreditCoinSellerStock(ctx, &walletv1.AdminCreditCoinSellerStockRequest{
		CommandId:        fmt.Sprintf("%s_stock", runID),
		SellerUserId:     sellerUserID,
		StockType:        "usdt_purchase",
		CoinAmount:       9_000_000,
		PaidCurrencyCode: "USDT",
		PaidAmountMicro:  100_000_000,
		PaymentRef:       fmt.Sprintf("%s_payment", runID),
		EvidenceRef:      "local-flow",
		OperatorUserId:   operatorAdminID,
		Reason:           "local cumulative recharge flow",
		AppCode:          appCode,
	})
	must("credit coin seller stock", err)
	transfer, err := client.TransferCoinFromSeller(ctx, &walletv1.TransferCoinFromSellerRequest{
		CommandId:      fmt.Sprintf("%s_seller_transfer", runID),
		SellerUserId:   sellerUserID,
		TargetUserId:   targetUserID,
		Amount:         9_000_000,
		Reason:         "local cumulative recharge flow",
		AppCode:        appCode,
		SellerRegionId: 1,
		TargetRegionId: 1,
	})
	must("transfer coin from seller", err)

	var fact walletOutboxFact
	err = db.QueryRowContext(ctx, `
		SELECT event_id, transaction_id, command_id, COALESCE(payload, '{}'), created_at_ms, available_delta
		FROM wallet_outbox
		WHERE app_code = ? AND transaction_id = ? AND user_id = ? AND event_type = 'WalletRechargeRecorded'
		LIMIT 1`,
		appCode, transfer.GetTransactionId(), targetUserID,
	).Scan(&fact.EventID, &fact.TransactionID, &fact.CommandID, &fact.PayloadJSON, &fact.CreatedAtMS, &fact.AvailableDelta)
	must("read coin seller WalletRechargeRecorded outbox", err)
	return fact
}

func setResourceGroupStatus(ctx context.Context, client walletv1.WalletServiceClient, runID string, groupID int64, status string) {
	_, err := client.SetResourceGroupStatus(ctx, &walletv1.SetResourceGroupStatusRequest{
		RequestId:      fmt.Sprintf("%s_group_status_%d_%s", runID, groupID, status),
		AppCode:        appCode,
		GroupId:        groupID,
		Status:         status,
		OperatorUserId: operatorAdminID,
	})
	must("set resource group status "+status, err)
}

func getStatus(ctx context.Context, client activityv1.CumulativeRechargeRewardServiceClient, userID int64) *activityv1.CumulativeRechargeRewardStatus {
	resp, err := client.GetCumulativeRechargeRewardStatus(ctx, &activityv1.GetCumulativeRechargeRewardStatusRequest{Meta: meta("get_status"), UserId: userID})
	must("get cumulative recharge status", err)
	return resp.GetStatus()
}

func runBoundaryFlow(ctx context.Context, walletClient walletv1.WalletServiceClient, rewardClient activityv1.CumulativeRechargeRewardServiceClient, adminClient activityv1.AdminCumulativeRechargeRewardServiceClient, activityDB *sql.DB, walletDB *sql.DB) {
	runID := fmt.Sprintf("crr_boundary_%d", time.Now().UnixMilli())
	baseUserID := int64(761000000000) + time.Now().Unix()%10000000
	now := time.Now().UTC()
	currentCycle := cycleForTime(now)
	nextCycle := cycleForTime(time.UnixMilli(currentCycle.EndMS).UTC().Add(time.Millisecond))

	group100 := createCoinRewardGroup(ctx, walletClient, runID, "100", 1000)
	group200 := createCoinRewardGroup(ctx, walletClient, runID, "200", 2000)

	updateConfig(ctx, adminClient, runID, group100, group200)
	duplicate := consumeRecharge(ctx, rewardClient, runID, "duplicate", baseUserID+1, "google_play", 13_500_000, 15_000, now.Add(10*time.Minute))
	assertGrantThresholds("duplicate first event should grant $100", duplicate.GetGrants(), []int64{10_000})
	duplicateAgain := consumeRecharge(ctx, rewardClient, runID, "duplicate", baseUserID+1, "google_play", 13_500_000, 15_000, now.Add(10*time.Minute))
	assert(duplicateAgain.GetReason() == "already_consumed" && len(duplicateAgain.GetGrants()) == 0, "duplicate event should be consumed without new grants")
	assert(queryProgress(ctx, activityDB, baseUserID+1, currentCycle.Key) == 15_000, "duplicate event must not double progress")
	assert(queryGrantSummary(ctx, activityDB, baseUserID+1, currentCycle.Key).Count == 1, "duplicate event must not double grant rows")

	cross := consumeRecharge(ctx, rewardClient, runID, "cross-multiple", baseUserID+2, "mifapay", 22_500_000, 25_000, now.Add(11*time.Minute))
	assertGrantThresholds("single large recharge should cross both tiers", cross.GetGrants(), []int64{10_000, 20_000})
	assert(queryProgress(ctx, activityDB, baseUserID+2, currentCycle.Key) == 25_000, "cross-tier progress should be $250.00")
	assert(queryWalletRewardSummary(ctx, walletDB, baseUserID+2).Count == 2, "cross-tier wallet should receive two reward grants")

	_, err := updateConfigWithTiers(ctx, adminClient, runID+"_duplicate_threshold", true, []*activityv1.CumulativeRechargeRewardTier{
		{TierCode: runID + "_dup_a", TierName: "Duplicate A", ThresholdUsdMinor: 10_000, ResourceGroupId: group100, Status: statusActive, SortOrder: 1},
		{TierCode: runID + "_dup_b", TierName: "Duplicate B", ThresholdUsdMinor: 10_000, ResourceGroupId: group200, Status: statusActive, SortOrder: 2},
	})
	assert(err != nil, "duplicate active thresholds must be rejected")

	_, err = updateConfigWithTiers(ctx, adminClient, runID+"_disabled", false, []*activityv1.CumulativeRechargeRewardTier{
		{TierCode: runID + "_disabled_100", TierName: "Disabled $100", ThresholdUsdMinor: 10_000, ResourceGroupId: group100, Status: statusActive, SortOrder: 1},
		{TierCode: runID + "_disabled_200", TierName: "Disabled $200", ThresholdUsdMinor: 20_000, ResourceGroupId: group200, Status: statusActive, SortOrder: 2},
	})
	must("disable cumulative recharge config", err)
	disabled := consumeRecharge(ctx, rewardClient, runID, "disabled-event", baseUserID+3, "google_play", 22_500_000, 25_000, now.Add(12*time.Minute))
	assert(disabled.GetReason() == "disabled" && len(disabled.GetGrants()) == 0, "disabled config should consume event without grants")
	assert(queryProgress(ctx, activityDB, baseUserID+3, currentCycle.Key) == 0, "disabled config should not update progress")
	updateConfig(ctx, adminClient, runID+"_reenable_after_disabled", group100, group200)
	disabledAgain := consumeRecharge(ctx, rewardClient, runID, "disabled-event", baseUserID+3, "google_play", 22_500_000, 25_000, now.Add(12*time.Minute))
	assert(disabledAgain.GetReason() == "already_consumed" && queryProgress(ctx, activityDB, baseUserID+3, currentCycle.Key) == 0, "disabled-period event must not backfill after re-enable")

	_, err = updateConfigWithTiers(ctx, adminClient, runID+"_inactive", true, []*activityv1.CumulativeRechargeRewardTier{
		{TierCode: runID + "_inactive_100", TierName: "Inactive $100", ThresholdUsdMinor: 10_000, ResourceGroupId: group100, Status: tierStatusInactive, SortOrder: 1},
		{TierCode: runID + "_active_200", TierName: "Active $200", ThresholdUsdMinor: 20_000, ResourceGroupId: group200, Status: statusActive, SortOrder: 2},
	})
	must("set inactive tier config", err)
	inactive := consumeRecharge(ctx, rewardClient, runID, "inactive-tier", baseUserID+4, "mifapay", 22_500_000, 25_000, now.Add(13*time.Minute))
	assertGrantThresholds("inactive $100 tier should be ignored", inactive.GetGrants(), []int64{20_000})
	assert(queryGrantSummary(ctx, activityDB, baseUserID+4, currentCycle.Key).Count == 1, "inactive tier should not create a grant row")

	failingGroup := createCoinRewardGroup(ctx, walletClient, runID, "disabled_group", 1000)
	setResourceGroupStatus(ctx, walletClient, runID, failingGroup, statusDisabled)
	_, err = updateConfigWithTiers(ctx, adminClient, runID+"_failed_retry", true, []*activityv1.CumulativeRechargeRewardTier{
		{TierCode: runID + "_failed_100", TierName: "Failed then retry", ThresholdUsdMinor: 10_000, ResourceGroupId: failingGroup, Status: statusActive, SortOrder: 1},
	})
	must("set disabled resource group tier", err)
	_, err = consumeRechargeAllowError(ctx, rewardClient, runID, "failed-wallet", baseUserID+5, "google_play", 13_500_000, 15_000, now.Add(14*time.Minute))
	assert(err != nil, "disabled wallet resource group should fail grant")
	assert(queryGrantStatus(ctx, activityDB, baseUserID+5, currentCycle.Key, 10_000) == "failed", "failed wallet grant should persist failed status")
	setResourceGroupStatus(ctx, walletClient, runID, failingGroup, statusActive)
	retried := consumeRecharge(ctx, rewardClient, runID, "failed-wallet", baseUserID+5, "google_play", 13_500_000, 15_000, now.Add(14*time.Minute))
	assertGrantThresholds("same event retry should grant after group enabled", retried.GetGrants(), []int64{10_000})
	assert(queryProgress(ctx, activityDB, baseUserID+5, currentCycle.Key) == 15_000, "failed retry must not double progress")
	assert(queryGrantStatus(ctx, activityDB, baseUserID+5, currentCycle.Key, 10_000) == "granted", "retried wallet grant should be granted")

	updateConfig(ctx, adminClient, runID+"_week_boundary", group100, group200)
	endBoundary := consumeRecharge(ctx, rewardClient, runID, "week-end", baseUserID+6, "google_play", 9_000_000, 10_000, time.UnixMilli(currentCycle.EndMS).UTC())
	assertGrantThresholds("event at cycle end should belong current cycle", endBoundary.GetGrants(), []int64{10_000})
	nextBoundary := consumeRecharge(ctx, rewardClient, runID, "week-next", baseUserID+6, "google_play", 9_000_000, 10_000, time.UnixMilli(currentCycle.EndMS+1).UTC())
	assertGrantThresholds("event after cycle end should belong next cycle", nextBoundary.GetGrants(), []int64{10_000})
	assert(queryProgress(ctx, activityDB, baseUserID+6, currentCycle.Key) == 10_000, "cycle end progress should stay current week")
	assert(queryProgress(ctx, activityDB, baseUserID+6, nextCycle.Key) == 10_000, "cycle start progress should reset next week")

	_, err = consumeRechargeAllowError(ctx, rewardClient, runID, "zero-usd", baseUserID+7, "google_play", 10_000, 0, now.Add(15*time.Minute))
	assert(err != nil, "zero qualifying USD event must be rejected by direct consume entry")
	assert(queryProgress(ctx, activityDB, baseUserID+7, currentCycle.Key) == 0, "zero qualifying USD event must not write progress")

	updateConfig(ctx, adminClient, runID+"_restore", group100, group200)
	out := map[string]any{
		"mode":          "boundary",
		"run_id":        runID,
		"base_user_id":  baseUserID,
		"current_cycle": currentCycle.Key,
		"next_cycle":    nextCycle.Key,
		"cases": map[string]any{
			"duplicate_event":       map[string]any{"progress_usd_minor": queryProgress(ctx, activityDB, baseUserID+1, currentCycle.Key), "grant_count": queryGrantSummary(ctx, activityDB, baseUserID+1, currentCycle.Key).Count},
			"cross_multiple_tiers":  map[string]any{"grant_thresholds": thresholds(cross.GetGrants()), "wallet_grants": queryWalletRewardSummary(ctx, walletDB, baseUserID+2).Count},
			"disabled_no_backfill":  map[string]any{"first_reason": disabled.GetReason(), "retry_reason": disabledAgain.GetReason(), "progress_usd_minor": queryProgress(ctx, activityDB, baseUserID+3, currentCycle.Key)},
			"inactive_tier_ignored": map[string]any{"grant_thresholds": thresholds(inactive.GetGrants())},
			"failed_retry":          map[string]any{"final_status": queryGrantStatus(ctx, activityDB, baseUserID+5, currentCycle.Key, 10_000), "progress_usd_minor": queryProgress(ctx, activityDB, baseUserID+5, currentCycle.Key)},
			"week_boundary":         map[string]any{"current_progress": queryProgress(ctx, activityDB, baseUserID+6, currentCycle.Key), "next_progress": queryProgress(ctx, activityDB, baseUserID+6, nextCycle.Key)},
			"zero_usd_rejected":     map[string]any{"progress_usd_minor": queryProgress(ctx, activityDB, baseUserID+7, currentCycle.Key)},
		},
	}
	encoded, _ := json.MarshalIndent(out, "", "  ")
	fmt.Println(string(encoded))
}

func queryProgress(ctx context.Context, db *sql.DB, userID int64, cycleKey string) int64 {
	var total sql.NullInt64
	err := db.QueryRowContext(ctx, `
		SELECT total_usd_minor
		FROM cumulative_recharge_reward_progress
		WHERE app_code = ? AND user_id = ? AND cycle_key = ?`,
		appCode, userID, cycleKey,
	).Scan(&total)
	if err == sql.ErrNoRows {
		return 0
	}
	must("query cumulative progress "+cycleKey, err)
	return total.Int64
}

type grantSummary struct {
	Count   int64 `json:"count"`
	Granted int64 `json:"granted"`
}

func queryGrantSummary(ctx context.Context, db *sql.DB, userID int64, cycleKey string) grantSummary {
	var summary grantSummary
	err := db.QueryRowContext(ctx, `
		SELECT COUNT(*), COALESCE(SUM(CASE WHEN status = 'granted' THEN 1 ELSE 0 END), 0)
		FROM cumulative_recharge_reward_grants
		WHERE app_code = ? AND user_id = ? AND cycle_key = ?`,
		appCode, userID, cycleKey,
	).Scan(&summary.Count, &summary.Granted)
	must("query cumulative grants "+cycleKey, err)
	return summary
}

func queryGrantStatus(ctx context.Context, db *sql.DB, userID int64, cycleKey string, thresholdUSDMinor int64) string {
	var status string
	err := db.QueryRowContext(ctx, `
		SELECT status
		FROM cumulative_recharge_reward_grants
		WHERE app_code = ? AND user_id = ? AND cycle_key = ? AND threshold_usd_minor = ?
		ORDER BY created_at_ms DESC
		LIMIT 1`,
		appCode, userID, cycleKey, thresholdUSDMinor,
	).Scan(&status)
	must("query cumulative grant status", err)
	return status
}

type walletRewardSummary struct {
	Count     int64 `json:"count"`
	CoinDelta int64 `json:"coin_delta"`
}

func queryWalletRewardSummary(ctx context.Context, db *sql.DB, userID int64) walletRewardSummary {
	var summary walletRewardSummary
	err := db.QueryRowContext(ctx, `
		SELECT COUNT(DISTINCT rg.grant_id), COALESCE(SUM(we.available_delta), 0)
		FROM resource_grants rg
		JOIN resource_grant_items rgi ON rgi.app_code = rg.app_code AND rgi.grant_id = rg.grant_id
		JOIN wallet_entries we ON we.app_code = rgi.app_code AND we.transaction_id = rgi.wallet_transaction_id AND we.user_id = rg.target_user_id
		WHERE rg.app_code = ? AND rg.target_user_id = ? AND rg.grant_source = ? AND we.asset_type = ?`,
		appCode, userID, rechargeSource, assetCoin,
	).Scan(&summary.Count, &summary.CoinDelta)
	must("query wallet reward grants", err)
	return summary
}

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

func cycleForTime(value time.Time) cycle {
	utc := value.UTC()
	dayStart := time.Date(utc.Year(), utc.Month(), utc.Day(), 0, 0, 0, 0, time.UTC)
	daysSinceMonday := (int(dayStart.Weekday()) + 6) % 7
	start := dayStart.AddDate(0, 0, -daysSinceMonday)
	end := start.AddDate(0, 0, 7)
	isoYear, isoWeek := start.ISOWeek()
	return cycle{Key: fmt.Sprintf("%04d-W%02d", isoYear, isoWeek), StartMS: start.UnixMilli(), EndMS: end.UnixMilli() - 1}
}

func assertGrantThresholds(message string, grants []*activityv1.CumulativeRechargeRewardGrant, want []int64) {
	got := thresholds(grants)
	if len(got) != len(want) {
		log.Fatalf("%s: got %v want %v", message, got, want)
	}
	for i := range want {
		if got[i] != want[i] {
			log.Fatalf("%s: got %v want %v", message, got, want)
		}
	}
}

func thresholds(grants []*activityv1.CumulativeRechargeRewardGrant) []int64 {
	items := make([]int64, 0, len(grants))
	for _, grant := range grants {
		if grant == nil {
			continue
		}
		items = append(items, grant.GetThresholdUsdMinor())
	}
	sort.Slice(items, func(i, j int) bool { return items[i] < items[j] })
	return items
}

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

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