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) } }