474 lines
19 KiB
Go
474 lines
19 KiB
Go
package main
|
||
|
||
import (
|
||
"context"
|
||
"database/sql"
|
||
"encoding/json"
|
||
"flag"
|
||
"fmt"
|
||
"log"
|
||
"os"
|
||
"sort"
|
||
"strings"
|
||
"time"
|
||
|
||
_ "github.com/go-sql-driver/mysql"
|
||
"google.golang.org/grpc"
|
||
"google.golang.org/grpc/credentials/insecure"
|
||
"google.golang.org/protobuf/proto"
|
||
activityv1 "hyapp.local/api/proto/activity/v1"
|
||
roomeventsv1 "hyapp.local/api/proto/events/room/v1"
|
||
)
|
||
|
||
const (
|
||
defaultAppCode = "lalu"
|
||
localTitlePrefix = "Local Weekly Star Flow"
|
||
activityCode = "weekly_star"
|
||
cycleStatusActive = "active"
|
||
rewardGroupTop1ID = int64(930001)
|
||
rewardGroupTop2ID = int64(930002)
|
||
rewardGroupTop3ID = int64(930003)
|
||
testOperatorAdmin = int64(900001)
|
||
testTargetAnchorID = int64(990001)
|
||
)
|
||
|
||
type flowConfig struct {
|
||
activityAddr string
|
||
activityDSN string
|
||
walletDSN string
|
||
appCode string
|
||
regionID int64
|
||
}
|
||
|
||
type flowSummary struct {
|
||
RunID string `json:"run_id"`
|
||
RegionID int64 `json:"region_id"`
|
||
FinishedCycleID string `json:"finished_cycle_id"`
|
||
CurrentCycleID string `json:"current_cycle_id"`
|
||
SentEvents []string `json:"sent_events"`
|
||
Leaderboard []leaderboardSummary `json:"leaderboard"`
|
||
Cron cronSummary `json:"cron"`
|
||
Settlements []settlementSummary `json:"settlements"`
|
||
WalletGrants []walletGrantSummary `json:"wallet_grants"`
|
||
CurrentCycleAfter string `json:"current_cycle_after"`
|
||
ExpectedAssertions map[string]string `json:"expected_assertions"`
|
||
}
|
||
|
||
type leaderboardSummary struct {
|
||
RankNo int32 `json:"rank_no"`
|
||
UserID int64 `json:"user_id"`
|
||
Score int64 `json:"score"`
|
||
}
|
||
|
||
type cronSummary struct {
|
||
Claimed int32 `json:"claimed"`
|
||
Processed int32 `json:"processed"`
|
||
Success int32 `json:"success"`
|
||
Failure int32 `json:"failure"`
|
||
}
|
||
|
||
type settlementSummary struct {
|
||
RankNo int32 `json:"rank_no"`
|
||
UserID int64 `json:"user_id"`
|
||
Score int64 `json:"score"`
|
||
ResourceGroupID int64 `json:"resource_group_id"`
|
||
WalletCommandID string `json:"wallet_command_id"`
|
||
WalletGrantID string `json:"wallet_grant_id"`
|
||
Status string `json:"status"`
|
||
}
|
||
|
||
type walletGrantSummary struct {
|
||
CommandID string `json:"command_id"`
|
||
GrantID string `json:"grant_id"`
|
||
TargetUserID int64 `json:"target_user_id"`
|
||
GrantSubjectID string `json:"grant_subject_id"`
|
||
Status string `json:"status"`
|
||
}
|
||
|
||
func main() {
|
||
cfg := flowConfig{}
|
||
flag.StringVar(&cfg.activityAddr, "activity-addr", "127.0.0.1:13006", "activity-service gRPC address")
|
||
flag.StringVar(&cfg.activityDSN, "activity-dsn", "hyapp:hyapp@tcp(127.0.0.1:23306)/hyapp_activity?parseTime=true&charset=utf8mb4&loc=UTC", "activity MySQL DSN")
|
||
flag.StringVar(&cfg.walletDSN, "wallet-dsn", "hyapp:hyapp@tcp(127.0.0.1:23306)/hyapp_wallet?parseTime=true&charset=utf8mb4&loc=UTC", "wallet MySQL DSN")
|
||
flag.StringVar(&cfg.appCode, "app-code", defaultAppCode, "app code")
|
||
flag.Int64Var(&cfg.regionID, "region-id", 990686, "local test region id")
|
||
flag.Parse()
|
||
if strings.TrimSpace(cfg.appCode) == "" {
|
||
log.Fatal("app-code is required")
|
||
}
|
||
|
||
ctx, cancel := context.WithTimeout(context.Background(), 40*time.Second)
|
||
defer cancel()
|
||
|
||
// activityDB 只用于准备和交叉验证测试夹具，不能直接写积分、settlement 或周期结果。
|
||
// 真正的周期创建、送礼消费、榜单读取和结算都必须走 activity-service gRPC，才能覆盖线上同一套业务逻辑。
|
||
activityDB := mustOpenDB(ctx, cfg.activityDSN)
|
||
defer activityDB.Close()
|
||
// walletDB 只用于准备本地资源组和最终审计 wallet grant；奖励发放必须通过 wallet-service GrantResourceGroup。
|
||
walletDB := mustOpenDB(ctx, cfg.walletDSN)
|
||
defer walletDB.Close()
|
||
|
||
// 只清理本工具创建的本地区域未完成周期，避免旧失败周期被本轮 cron 一起重试。
|
||
mustExec(ctx, activityDB, `
|
||
UPDATE weekly_star_cycles
|
||
SET status = 'disabled', locked_by = '', lock_until_ms = 0, updated_at_ms = ?
|
||
WHERE app_code = ? AND activity_code = ? AND region_id = ? AND title LIKE ? AND status IN ('active', 'settling')`,
|
||
time.Now().UTC().UnixMilli(), cfg.appCode, activityCode, cfg.regionID, localTitlePrefix+"%",
|
||
)
|
||
// 奖励资源组由 wallet-service 拥有；这里仅准备本地验证用的 COIN 资源组配置。
|
||
seedWalletRewardGroups(ctx, walletDB, cfg.appCode)
|
||
|
||
conn, err := grpc.DialContext(ctx, cfg.activityAddr, grpc.WithTransportCredentials(insecure.NewCredentials()), grpc.WithBlock())
|
||
if err != nil {
|
||
log.Fatalf("dial activity-service: %v", err)
|
||
}
|
||
defer conn.Close()
|
||
|
||
adminClient := activityv1.NewAdminWeeklyStarServiceClient(conn)
|
||
appClient := activityv1.NewWeeklyStarServiceClient(conn)
|
||
roomClient := activityv1.NewRoomEventConsumerServiceClient(conn)
|
||
cronClient := activityv1.NewActivityCronServiceClient(conn)
|
||
|
||
runID := fmt.Sprintf("%d", time.Now().UTC().UnixMilli())
|
||
now := time.Now().UTC()
|
||
// 第一个周期刻意设置为已经结束，事件 occurred_at_ms 落在它的 [start,end) 内，cron 可以立即结算。
|
||
// 第二个周期从第一个周期 end_ms 开始覆盖当前时间，用来验证旧周期结算后 current 自动进入下一周期。
|
||
finishedStart := now.Add(-10 * time.Minute).UnixMilli()
|
||
finishedEnd := now.Add(-1 * time.Minute).UnixMilli()
|
||
currentStart := finishedEnd
|
||
currentEnd := now.Add(50 * time.Minute).UnixMilli()
|
||
gifts := []string{"weekly_star_flow_gift_a", "weekly_star_flow_gift_b", "weekly_star_flow_gift_c"}
|
||
|
||
// 后台配置两个不重叠周期：第一个已经到期用于结算，第二个覆盖当前时间用于验证自动进入下一周期。
|
||
finishedCycle := createCycle(ctx, adminClient, cfg, fmt.Sprintf("%s finished %s", localTitlePrefix, runID), finishedStart, finishedEnd, gifts)
|
||
currentCycle := createCycle(ctx, adminClient, cfg, fmt.Sprintf("%s current %s", localTitlePrefix, runID), currentStart, currentEnd, gifts)
|
||
|
||
eventTime := finishedEnd - 10_000
|
||
events := []struct {
|
||
userID int64
|
||
giftID string
|
||
score int64
|
||
}{
|
||
{userID: 701001, giftID: gifts[0], score: 5_000},
|
||
{userID: 701002, giftID: gifts[1], score: 8_000},
|
||
{userID: 701003, giftID: gifts[2], score: 6_500},
|
||
}
|
||
sentEventIDs := make([]string, 0, len(events)+1)
|
||
for index, event := range events {
|
||
eventID := fmt.Sprintf("weekly-star-flow-%s-%d", runID, index+1)
|
||
// consumeGiftEvent 调的是 RoomEventConsumerService，不直接调用 weekly-star service。
|
||
// 这样可以同时验证 activity-service 的房间事件 fanout wiring 是否把周星模块挂上。
|
||
consumeGiftEvent(ctx, roomClient, cfg, eventID, event.userID, event.giftID, event.score, eventTime, int64(index+1))
|
||
sentEventIDs = append(sentEventIDs, eventID)
|
||
}
|
||
// 同一个 source_event_id 再投一次，验证 weekly_star_score_events 的幂等约束不会让榜单重复加分。
|
||
consumeGiftEvent(ctx, roomClient, cfg, sentEventIDs[0], events[0].userID, events[0].giftID, events[0].score, eventTime, 99)
|
||
sentEventIDs = append(sentEventIDs, sentEventIDs[0]+" (duplicate)")
|
||
|
||
leaderboard := listLeaderboard(ctx, adminClient, cfg, finishedCycle.GetCycleId())
|
||
// 先查榜再结算，验证积分累计发生在送礼事件消费阶段，而不是结算时临时计算。
|
||
assertLeaderboard(leaderboard, []leaderboardSummary{
|
||
{RankNo: 1, UserID: 701002, Score: 8_000},
|
||
{RankNo: 2, UserID: 701003, Score: 6_500},
|
||
{RankNo: 3, UserID: 701001, Score: 5_000},
|
||
})
|
||
|
||
cronResp, err := cronClient.ProcessWeeklyStarSettlementBatch(ctx, &activityv1.CronBatchRequest{
|
||
Meta: meta(cfg, "weekly-star-local-flow-cron-"+runID),
|
||
RunId: "weekly-star-local-flow-" + runID,
|
||
WorkerId: "weekly-star-local-flow",
|
||
BatchSize: 10,
|
||
LockTtlMs: int64((30 * time.Second) / time.Millisecond),
|
||
})
|
||
if err != nil {
|
||
log.Fatalf("process weekly star settlement: %v", err)
|
||
}
|
||
if cronResp.GetClaimedCount() != 1 || cronResp.GetProcessedCount() != 3 || cronResp.GetSuccessCount() != 3 || cronResp.GetFailureCount() != 0 {
|
||
log.Fatalf("unexpected settlement cron result: claimed=%d processed=%d success=%d failure=%d", cronResp.GetClaimedCount(), cronResp.GetProcessedCount(), cronResp.GetSuccessCount(), cronResp.GetFailureCount())
|
||
}
|
||
|
||
settlements := listSettlements(ctx, adminClient, cfg, finishedCycle.GetCycleId())
|
||
assertSettlements(settlements)
|
||
// activity 侧 granted 只能说明它收到了 wallet 返回；这里再查 wallet.resource_grants，确认奖励确实落到 wallet owner 库。
|
||
grants := listWalletGrants(ctx, walletDB, cfg.appCode, walletCommandIDs(settlements))
|
||
if len(grants) != 3 {
|
||
log.Fatalf("expected 3 wallet grants, got %d", len(grants))
|
||
}
|
||
|
||
currentResp, err := appClient.GetWeeklyStarCurrent(ctx, &activityv1.GetWeeklyStarCurrentRequest{
|
||
Meta: meta(cfg, "weekly-star-local-flow-current-"+runID),
|
||
UserId: 701001,
|
||
RegionId: cfg.regionID,
|
||
})
|
||
if err != nil {
|
||
log.Fatalf("get current weekly star cycle: %v", err)
|
||
}
|
||
if currentResp.GetCycle().GetCycleId() != currentCycle.GetCycleId() {
|
||
log.Fatalf("expected current cycle %s, got %s", currentCycle.GetCycleId(), currentResp.GetCycle().GetCycleId())
|
||
}
|
||
|
||
summary := flowSummary{
|
||
RunID: runID,
|
||
RegionID: cfg.regionID,
|
||
FinishedCycleID: finishedCycle.GetCycleId(),
|
||
CurrentCycleID: currentCycle.GetCycleId(),
|
||
SentEvents: sentEventIDs,
|
||
Leaderboard: leaderboard,
|
||
Cron: cronSummary{
|
||
Claimed: cronResp.GetClaimedCount(),
|
||
Processed: cronResp.GetProcessedCount(),
|
||
Success: cronResp.GetSuccessCount(),
|
||
Failure: cronResp.GetFailureCount(),
|
||
},
|
||
Settlements: settlements,
|
||
WalletGrants: grants,
|
||
CurrentCycleAfter: currentResp.GetCycle().GetCycleId(),
|
||
ExpectedAssertions: map[string]string{
|
||
"multiple_cycles": "created one ended active cycle and one current active cycle in UTC epoch ms",
|
||
"gift_score": "configured gift score equals RoomGiftSent.coin_spent",
|
||
"duplicate_event": "same source_event_id did not double-count leaderboard score",
|
||
"settlement": "ended cycle settled top1/top2/top3 through wallet GrantResourceGroup",
|
||
"next_cycle": "GetWeeklyStarCurrent returned the second cycle after settlement",
|
||
},
|
||
}
|
||
encoder := json.NewEncoder(os.Stdout)
|
||
encoder.SetIndent("", " ")
|
||
if err := encoder.Encode(summary); err != nil {
|
||
log.Fatalf("encode summary: %v", err)
|
||
}
|
||
}
|
||
|
||
func mustOpenDB(ctx context.Context, dsn string) *sql.DB {
|
||
db, err := sql.Open("mysql", dsn)
|
||
if err != nil {
|
||
log.Fatalf("open mysql: %v", err)
|
||
}
|
||
if err := db.PingContext(ctx); err != nil {
|
||
_ = db.Close()
|
||
log.Fatalf("ping mysql: %v", err)
|
||
}
|
||
return db
|
||
}
|
||
|
||
func mustExec(ctx context.Context, db *sql.DB, query string, args ...any) {
|
||
if _, err := db.ExecContext(ctx, query, args...); err != nil {
|
||
log.Fatalf("exec mysql setup: %v", err)
|
||
}
|
||
}
|
||
|
||
func seedWalletRewardGroups(ctx context.Context, db *sql.DB, appCode string) {
|
||
nowMS := time.Now().UTC().UnixMilli()
|
||
type groupSeed struct {
|
||
groupID int64
|
||
code string
|
||
name string
|
||
coins int64
|
||
}
|
||
seeds := []groupSeed{
|
||
{groupID: rewardGroupTop1ID, code: "weekly_star_local_top1", name: "Weekly Star Local Top 1", coins: 111_000},
|
||
{groupID: rewardGroupTop2ID, code: "weekly_star_local_top2", name: "Weekly Star Local Top 2", coins: 66_000},
|
||
{groupID: rewardGroupTop3ID, code: "weekly_star_local_top3", name: "Weekly Star Local Top 3", coins: 33_000},
|
||
}
|
||
for index, seed := range seeds {
|
||
// 固定 group_id 让 activity 周期配置可以稳定引用；ON DUPLICATE 只更新本工具自己的本地资源组。
|
||
mustExec(ctx, db, `
|
||
INSERT INTO resource_groups (
|
||
app_code, 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', 'local weekly star flow reward group', ?, ?, ?, ?, ?)
|
||
ON DUPLICATE KEY UPDATE
|
||
name = VALUES(name), status = 'active', description = VALUES(description),
|
||
sort_order = VALUES(sort_order), updated_by_user_id = VALUES(updated_by_user_id), updated_at_ms = VALUES(updated_at_ms)`,
|
||
appCode, seed.groupID, seed.code, seed.name, index+1, testOperatorAdmin, testOperatorAdmin, nowMS, nowMS,
|
||
)
|
||
mustExec(ctx, db, `
|
||
-- wallet_asset item 不依赖具体 resource_id，wallet-service 会把它原子展开成 COIN 入账和 resource_grant_item。
|
||
INSERT INTO resource_group_items (
|
||
app_code, group_id, item_type, resource_id, wallet_asset_type, wallet_asset_amount,
|
||
quantity, duration_ms, sort_order, created_at_ms, updated_at_ms
|
||
) VALUES (?, ?, 'wallet_asset', 0, 'COIN', ?, 1, 0, 1, ?, ?)
|
||
ON DUPLICATE KEY UPDATE
|
||
wallet_asset_amount = VALUES(wallet_asset_amount), sort_order = VALUES(sort_order), updated_at_ms = VALUES(updated_at_ms)`,
|
||
appCode, seed.groupID, seed.coins, nowMS, nowMS,
|
||
)
|
||
}
|
||
}
|
||
|
||
func createCycle(ctx context.Context, client activityv1.AdminWeeklyStarServiceClient, cfg flowConfig, title string, startMS int64, endMS int64, gifts []string) *activityv1.WeeklyStarCycle {
|
||
resp, err := client.CreateWeeklyStarCycle(ctx, &activityv1.UpsertWeeklyStarCycleRequest{
|
||
Meta: meta(cfg, "weekly-star-local-flow-create-"+strings.ReplaceAll(title, " ", "-")),
|
||
Cycle: &activityv1.WeeklyStarCycle{
|
||
ActivityCode: activityCode,
|
||
RegionId: cfg.regionID,
|
||
Title: title,
|
||
Status: cycleStatusActive,
|
||
StartMs: startMS,
|
||
EndMs: endMS,
|
||
Gifts: []*activityv1.WeeklyStarGift{
|
||
{GiftId: gifts[0], SortOrder: 1},
|
||
{GiftId: gifts[1], SortOrder: 2},
|
||
{GiftId: gifts[2], SortOrder: 3},
|
||
},
|
||
Rewards: []*activityv1.WeeklyStarReward{
|
||
{RankNo: 1, ResourceGroupId: rewardGroupTop1ID},
|
||
{RankNo: 2, ResourceGroupId: rewardGroupTop2ID},
|
||
{RankNo: 3, ResourceGroupId: rewardGroupTop3ID},
|
||
},
|
||
},
|
||
OperatorAdminId: testOperatorAdmin,
|
||
})
|
||
if err != nil {
|
||
log.Fatalf("create cycle %q: %v", title, err)
|
||
}
|
||
if resp.GetCycle().GetCycleId() == "" {
|
||
log.Fatalf("create cycle %q returned empty cycle_id", title)
|
||
}
|
||
return resp.GetCycle()
|
||
}
|
||
|
||
func consumeGiftEvent(ctx context.Context, client activityv1.RoomEventConsumerServiceClient, cfg flowConfig, eventID string, userID int64, giftID string, score int64, occurredAtMS int64, version int64) {
|
||
body, err := proto.Marshal(&roomeventsv1.RoomGiftSent{
|
||
SenderUserId: userID,
|
||
TargetUserId: testTargetAnchorID,
|
||
GiftId: giftID,
|
||
GiftCount: 1,
|
||
GiftValue: 1,
|
||
BillingReceiptId: eventID + "-receipt",
|
||
VisibleRegionId: cfg.regionID,
|
||
CommandId: eventID + "-command",
|
||
CoinSpent: score,
|
||
})
|
||
if err != nil {
|
||
log.Fatalf("marshal room gift sent: %v", err)
|
||
}
|
||
_, err = client.ConsumeRoomEvent(ctx, &activityv1.ConsumeRoomEventRequest{
|
||
Meta: meta(cfg, "weekly-star-local-flow-event-"+eventID),
|
||
Envelope: &roomeventsv1.EventEnvelope{
|
||
EventId: eventID,
|
||
RoomId: "weekly-star-local-room",
|
||
EventType: "RoomGiftSent",
|
||
RoomVersion: version,
|
||
OccurredAtMs: occurredAtMS,
|
||
Body: body,
|
||
AppCode: cfg.appCode,
|
||
},
|
||
})
|
||
if err != nil {
|
||
log.Fatalf("consume room gift event %s: %v", eventID, err)
|
||
}
|
||
}
|
||
|
||
func listLeaderboard(ctx context.Context, client activityv1.AdminWeeklyStarServiceClient, cfg flowConfig, cycleID string) []leaderboardSummary {
|
||
resp, err := client.ListWeeklyStarLeaderboard(ctx, &activityv1.ListWeeklyStarLeaderboardRequest{
|
||
Meta: meta(cfg, "weekly-star-local-flow-leaderboard-"+cycleID),
|
||
CycleId: cycleID,
|
||
RegionId: cfg.regionID,
|
||
PageSize: 10,
|
||
})
|
||
if err != nil {
|
||
log.Fatalf("list leaderboard: %v", err)
|
||
}
|
||
out := make([]leaderboardSummary, 0, len(resp.GetEntries()))
|
||
for _, entry := range resp.GetEntries() {
|
||
out = append(out, leaderboardSummary{RankNo: entry.GetRankNo(), UserID: entry.GetUserId(), Score: entry.GetScore()})
|
||
}
|
||
return out
|
||
}
|
||
|
||
func assertLeaderboard(actual []leaderboardSummary, expected []leaderboardSummary) {
|
||
if len(actual) < len(expected) {
|
||
log.Fatalf("expected at least %d leaderboard entries, got %d", len(expected), len(actual))
|
||
}
|
||
for index, want := range expected {
|
||
got := actual[index]
|
||
if got != want {
|
||
log.Fatalf("leaderboard row %d mismatch: got %+v want %+v", index+1, got, want)
|
||
}
|
||
}
|
||
}
|
||
|
||
func listSettlements(ctx context.Context, client activityv1.AdminWeeklyStarServiceClient, cfg flowConfig, cycleID string) []settlementSummary {
|
||
resp, err := client.ListWeeklyStarSettlements(ctx, &activityv1.ListWeeklyStarSettlementsRequest{
|
||
Meta: meta(cfg, "weekly-star-local-flow-settlements-"+cycleID),
|
||
CycleId: cycleID,
|
||
})
|
||
if err != nil {
|
||
log.Fatalf("list settlements: %v", err)
|
||
}
|
||
out := make([]settlementSummary, 0, len(resp.GetSettlements()))
|
||
for _, settlement := range resp.GetSettlements() {
|
||
out = append(out, settlementSummary{
|
||
RankNo: settlement.GetRankNo(),
|
||
UserID: settlement.GetUserId(),
|
||
Score: settlement.GetScore(),
|
||
ResourceGroupID: settlement.GetResourceGroupId(),
|
||
WalletCommandID: settlement.GetWalletCommandId(),
|
||
WalletGrantID: settlement.GetWalletGrantId(),
|
||
Status: settlement.GetStatus(),
|
||
})
|
||
}
|
||
sort.Slice(out, func(i, j int) bool { return out[i].RankNo < out[j].RankNo })
|
||
return out
|
||
}
|
||
|
||
func assertSettlements(items []settlementSummary) {
|
||
if len(items) != 3 {
|
||
log.Fatalf("expected 3 settlements, got %d", len(items))
|
||
}
|
||
for _, item := range items {
|
||
if item.Status != "granted" || item.WalletCommandID == "" || item.WalletGrantID == "" {
|
||
log.Fatalf("settlement was not granted: %+v", item)
|
||
}
|
||
}
|
||
}
|
||
|
||
func walletCommandIDs(items []settlementSummary) []string {
|
||
ids := make([]string, 0, len(items))
|
||
for _, item := range items {
|
||
ids = append(ids, item.WalletCommandID)
|
||
}
|
||
return ids
|
||
}
|
||
|
||
func listWalletGrants(ctx context.Context, db *sql.DB, appCode string, commandIDs []string) []walletGrantSummary {
|
||
if len(commandIDs) == 0 {
|
||
return nil
|
||
}
|
||
placeholders := strings.TrimRight(strings.Repeat("?,", len(commandIDs)), ",")
|
||
args := make([]any, 0, len(commandIDs)+1)
|
||
args = append(args, appCode)
|
||
for _, id := range commandIDs {
|
||
args = append(args, id)
|
||
}
|
||
rows, err := db.QueryContext(ctx, `
|
||
SELECT command_id, grant_id, target_user_id, grant_subject_id, status
|
||
FROM resource_grants
|
||
WHERE app_code = ? AND command_id IN (`+placeholders+`)
|
||
ORDER BY target_user_id ASC`, args...)
|
||
if err != nil {
|
||
log.Fatalf("query wallet grants: %v", err)
|
||
}
|
||
defer rows.Close()
|
||
out := make([]walletGrantSummary, 0, len(commandIDs))
|
||
for rows.Next() {
|
||
var item walletGrantSummary
|
||
if err := rows.Scan(&item.CommandID, &item.GrantID, &item.TargetUserID, &item.GrantSubjectID, &item.Status); err != nil {
|
||
log.Fatalf("scan wallet grant: %v", err)
|
||
}
|
||
out = append(out, item)
|
||
}
|
||
if err := rows.Err(); err != nil {
|
||
log.Fatalf("iterate wallet grants: %v", err)
|
||
}
|
||
return out
|
||
}
|
||
|
||
func meta(cfg flowConfig, requestID string) *activityv1.RequestMeta {
|
||
return &activityv1.RequestMeta{
|
||
RequestId: requestID,
|
||
Caller: "weekly-star-local-flow",
|
||
SentAtMs: time.Now().UTC().UnixMilli(),
|
||
AppCode: cfg.appCode,
|
||
}
|
||
}
|