zerosaturation 0d6b7a9e95 feat:增加购买道具的聚合功能

2026-06-24 19:51:13 +08:00

37 KiB

Raw Blame History

用户贡献连击聚合推送 — 实施计划

For agentic workers: REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (- [ ]) syntax for tracking.

Goal: 同用户 3 秒内连续送同一道具 N 次时,后端只推送 1 条合并 record,前端只显示 1 条 UI 项,数量 = N。

Architecture: 后端将 Redis combo:{userID}:{itemType} 计数器改造为 Redis Stream 延迟队列 + 聚合 Hash,worker 协程 3 秒后从 Stream 取出条目、读聚合 Hash、Publish 1 条合并 record。GetLatestContributions 在 SQL 后做内存合并。前端 WS 与轮询两条路径共用 mergeComboRecords 工具函数。

Tech Stack: Go 1.21+ (go-redis/v9 Stream API), Vue 3 Composition API, uni-app, Jest (前端单测)/ stretchr/testify (后端单测)

Working Branch: feat/actvity (当前分支,无需切换)

Spec: docs/superpowers/specs/2026-06-23-contribution-combo-aggregation-design.md

File Structure

后端 (Go)

文件	责任
`backend/services/activityService/service/combo_worker.go` (新建)	Stream 消费者 goroutine,负责 3 秒窗口结束后从 Stream 取出条目 + 读聚合 Hash + Publish 合并 record
`backend/services/activityService/service/activity_service.go` (修改)	改造 `PurchaseItem` / `BatchPurchaseItem`:用 agg Hash + XAdd 替代 `incrementComboCount` + 立即 Publish;改造 `GetLatestContributions`:SQL 后做内存合并;删除 `incrementComboCount` / `getComboCount`;保留 `comboKey` (供测试或其他辅助调用)
`backend/services/activityService/main.go` (修改)	在 `NewActivityService` 之后启动 combo worker goroutine

前端 (Vue / uni-app)

文件	责任
`frontend/pages/support-activity/composables/useContributionRealtime.js` (修改)	导出 `mergeComboRecords` 工具函数;`onWsMessage` 调用合并
`frontend/pages/support-activity/composables/useContributionPolling.js` (修改)	在 `fetchLatest` 合并新记录后调用 `mergeComboRecords`
`frontend/pages/support-activity/components/ContributionList.vue` (核对/微调)	模板已用 `record.quantity`(spec §5.4 "Before" 与现状不一致);核对后若已正确则跳过,若仍用 `combo_count` 则改为 `quantity`

测试

文件	责任
`backend/services/activityService/service/combo_worker_test.go` (新建)	`mergeComboAggHash` / `processComboEntry` 单元测试
`backend/services/activityService/service/activity_service_combo_test.go` (新建)	`PurchaseItem` Redis 不可用降级 / `GetLatestContributions` 内存合并单元测试
`frontend/pages/support-activity/composables/useContributionRealtime.test.js` (新建,可选)	`mergeComboRecords` 工具函数单测

任务清单

Task 1: 后端 — 删除旧的 `incrementComboCount` 与 `getComboCount`,新增聚合键生成函数

Files:

Modify: backend/services/activityService/service/activity_service.go:111-145
Step 1.1: 阅读并确认当前 combo 代码位置

grep -n "incrementComboCount\|getComboCount\|comboKey" \
  backend/services/activityService/service/activity_service.go

预期输出应包含:

第 111 行: comboKey 函数
第 118 行: incrementComboCount
第 135 行: getComboCount
Step 1.2: 删除 incrementComboCount 与 getComboCount

打开 backend/services/activityService/service/activity_service.go,定位到第 116-145 行(即两个 // ... 函数体),删除 整个函数块 但保留 comboKey(供测试/调试用)。删除后文件中应不再出现 incrementComboCount 或 getComboCount 字样。

Step 1.3: 新增 aggKey 与 aggLockKey 键生成函数

在 comboKey 函数下方(第 114 行 } 之后)新增:

// aggKey 生成聚合 Hash Redis Key（窗口内累加 quantity 等字段）
func (s *activityService) aggKey(userID int64, itemType string) string {
    return fmt.Sprintf("combo:agg:%d:%s", userID, itemType)
}

// aggLockKey 生成"是否已写入 Stream"防重锁 Key（SET NX EX 3s）
func (s *activityService) aggLockKey(userID int64, itemType string) string {
    return fmt.Sprintf("combo:agg:lock:%d:%s", userID, itemType)
}

Step 1.4: 验证编译

cd backend && go build ./services/activityService/...

预期: BUILD OK。注意此时 PurchaseItem/BatchPurchaseItem/GetLatestContributions 还在调用已被删除的 incrementComboCount/getComboCount,编译会失败 —— 这是预期的,Task 2/3/4 会修复。

Step 1.5: 暂不 commit

继续 Task 2。

Task 2: 后端 — 新增 `combo_worker.go`(Stream 消费者 + 聚合辅助函数)

Files:

Create: backend/services/activityService/service/combo_worker.go
Step 2.1: 创建 combo_worker.go 文件骨架

新建 backend/services/activityService/service/combo_worker.go,内容:

package service

import (
    "context"
    "encoding/json"
    "fmt"
    "strconv"
    "time"

    "github.com/redis/go-redis/v9"
    "github.com/topfans/backend/pkg/logger"
    pb "github.com/topfans/backend/pkg/proto/activity"
    "go.uber.org/zap"
)

const (
    comboStreamKey    = "combo:stream:contributions"
    comboConsumerGrp  = "combo-publishers"
    comboAggTTL       = 5 * time.Second
    comboWindowTTL    = 3 * time.Second
    comboStreamMaxLen = 100000
)

// StartComboStreamWorker 启动 Stream 消费者 goroutine
// 主进程启动后调用一次;通过 ctx 取消关闭
func (s *activityService) StartComboStreamWorker(ctx context.Context) {
    if s.redisClient == nil {
        logger.Logger.Warn("combo worker skipped: redisClient is nil")
        return
    }

    consumerName := fmt.Sprintf("worker-%d", time.Now().UnixNano()%100000)

    // 创建消费者组（已存在则忽略 BUSYGROUP）
    _ = s.redisClient.XGroupCreateMkStream(ctx, comboStreamKey, comboConsumerGrp, "0").Err()

    go func() {
        for {
            select {
            case <-ctx.Done():
                logger.Logger.Info("combo worker stopped")
                return
            default:
            }

            streams, err := s.redisClient.XReadGroup(ctx, &redis.XReadGroupArgs{
                Group:    comboConsumerGrp,
                Consumer: consumerName,
                Streams:  []string{comboStreamKey, ">"},
                Count:    10,
                Block:    1 * time.Second,
            }).Result()

            if err == redis.Nil {
                continue
            }
            if err != nil {
                if ctx.Err() != nil {
                    return
                }
                logger.Logger.Warn("combo XReadGroup error", zap.Error(err))
                time.Sleep(time.Second)
                continue
            }

            for _, stream := range streams {
                for _, msg := range stream.Messages {
                    s.processComboEntry(ctx, msg)
                    s.redisClient.XAck(ctx, comboStreamKey, comboConsumerGrp, msg.ID)
                }
            }
        }
    }()
}

// processComboEntry 处理一条 Stream 条目:
// - 等到 expire_at_ms 才推送（保证窗口结束）
// - 读取 agg Hash 推送合并 record
// - 推送后清理 agg + lock Key
func (s *activityService) processComboEntry(ctx context.Context, msg redis.XMessage) {
    expireAtRaw, ok := msg.Values["expire_at_ms"]
    if !ok {
        return
    }
    expireAtStr, ok := expireAtRaw.(string)
    if !ok {
        return
    }
    expireAt, err := strconv.ParseInt(expireAtStr, 10, 64)
    if err != nil {
        return
    }

    now := time.Now().UnixMilli()
    if now < expireAt {
        select {
        case <-ctx.Done():
            return
        case <-time.After(time.Duration(expireAt-now) * time.Millisecond):
        }
    }

    userIDStr, _ := msg.Values["user_id"].(string)
    activityIDStr, _ := msg.Values["activity_id"].(string)
    itemType, _ := msg.Values["item_type"].(string)
    userID, _ := strconv.ParseInt(userIDStr, 10, 64)
    activityID, _ := strconv.ParseInt(activityIDStr, 10, 64)
    if userID == 0 || activityID == 0 || itemType == "" {
        return
    }

    aggKey := s.aggKey(userID, itemType)
    lockKey := s.aggLockKey(userID, itemType)

    fields, err := s.redisClient.HGetAll(ctx, aggKey).Result()
    if err != nil || len(fields) == 0 {
        return
    }

    record := buildComboRecord(fields, userID, activityID, itemType)
    if record == nil {
        return
    }

    payload, _ := json.Marshal(map[string]interface{}{
        "activity_id": activityID,
        "type":        "contributions_response",
        "record":      record,
    })
    channel := fmt.Sprintf("act:%d:contributions", activityID)
    if err := s.redisClient.Publish(ctx, channel, payload).Err(); err != nil {
        logger.Logger.Warn("combo Publish failed", zap.Error(err))
        return // 不删 hash/lock,等下次重试
    }
    s.redisClient.Del(ctx, aggKey, lockKey)
}

// buildComboRecord 把 agg Hash 字段组装成 ContributionRecord
// 抽成独立函数便于单测;quantity 不解析成功时返回 nil
func buildComboRecord(fields map[string]string, userID, activityID int64, itemType string) *pb.ContributionRecord {
    quantity, err := strconv.ParseInt(fields["quantity"], 10, 32)
    if err != nil {
        return nil
    }
    firstID, err := strconv.ParseInt(fields["first_id"], 10, 64)
    if err != nil {
        return nil
    }
    firstCreatedAt, err := strconv.ParseInt(fields["first_created_at"], 10, 64)
    if err != nil {
        return nil
    }

    return &pb.ContributionRecord{
        Id:         firstID,
        UserId:     userID,
        Nickname:   fields["nickname"],
        AvatarUrl:  fields["avatar_url"],
        StarId:     parseInt64(fields["star_id"]),
        ItemId:     parseInt64(fields["item_id"]),
        ItemType:   itemType,
        ItemName:   fields["item_name"],
        ItemIcon:   fields["item_icon"],
        Quantity:   int32(quantity),
        ComboCount: int32(quantity),
        CreatedAt:  firstCreatedAt,
    }
}

func parseInt64(s string) int64 {
    n, _ := strconv.ParseInt(s, 10, 64)
    return n
}

Step 2.2: 写 buildComboRecord 单元测试

新建 backend/services/activityService/service/combo_worker_test.go:

package service

import (
    "testing"

    "github.com/stretchr/testify/assert"
)

func TestBuildComboRecord_Success(t *testing.T) {
    fields := map[string]string{
        "quantity":         "5",
        "first_id":         "1234",
        "first_created_at": "1700000000000",
        "nickname":         "小明",
        "avatar_url":       "https://cdn/x.jpg",
        "star_id":          "7",
        "item_id":          "42",
        "item_name":        "烟花",
        "item_icon":        "https://cdn/y.png",
    }
    r := buildComboRecord(fields, 100, 200, "firework")
    assert.NotNil(t, r)
    assert.Equal(t, int64(1234), r.Id)
    assert.Equal(t, int64(100), r.UserId)
    assert.Equal(t, "小明", r.Nickname)
    assert.Equal(t, "烟花", r.ItemName)
    assert.Equal(t, int32(5), r.Quantity)
    assert.Equal(t, int32(5), r.ComboCount)
    assert.Equal(t, int64(7), r.StarId)
    assert.Equal(t, int64(42), r.ItemId)
}

func TestBuildComboRecord_QuantityInvalid(t *testing.T) {
    fields := map[string]string{
        "quantity": "not-a-number",
        "first_id": "1",
    }
    assert.Nil(t, buildComboRecord(fields, 1, 1, "x"))
}

func TestBuildComboRecord_FirstIDMissing(t *testing.T) {
    fields := map[string]string{
        "quantity": "3",
    }
    assert.Nil(t, buildComboRecord(fields, 1, 1, "x"))
}

Step 2.3: 运行测试验证通过

cd backend && go test ./services/activityService/service/ -run TestBuildComboRecord -v

预期: 3 tests passed

Step 2.4: 暂不 commit

继续 Task 3。

Task 3: 后端 — 改造 `PurchaseItem` 使用 agg Hash + XAdd,Redis 不可用时降级

Files:

Modify: backend/services/activityService/service/activity_service.go:454-490
Step 3.1: 阅读当前 PurchaseItem 推送段

定位到 activity_service.go 第 454-490 行。整段包含 incrementComboCount → getComboCount → Publish 三步,需整体替换。

Step 3.2: 替换为聚合 + 入队逻辑

将第 454-490 行(从 // 更新 Redis 连击计数器 注释开始,到 s.redisClient.Publish(...) 结束)替换为:

    // 连击聚合:同 user+itemType 在 3 秒窗口内合并为 1 条 WS 推送
    // (1) 累加 agg Hash quantity (其他字段 HSetNX 保护首次信息)
    // (2) 仅首次入队 Stream (SET NX EX 3 防重复)
    // (3) Redis 不可用时降级为立即 Publish 单条 record
    nickname, avatarURL := "", ""
    if profile, _ := s.userRPCClient.GetFanProfile(userID, req.StarId); profile != nil {
        nickname = profile.Nickname
        avatarURL = profile.AvatarUrl
    }
    itemName, itemIcon := "", ""
    if item != nil {
        itemName = item.ItemName
        itemIcon = item.IconURL
    }

    s.enqueueComboContribution(ctx, comboEnqueueParams{
        ActivityID:      req.ActivityId,
        UserID:          userID,
        StarID:          req.StarId,
        ItemID:          item.ID,
        ItemType:        req.ItemType,
        ItemName:        itemName,
        ItemIcon:        itemIcon,
        Nickname:        nickname,
        AvatarURL:       avatarURL,
        FirstID:         contribution.ID,
        FirstCreatedAt:  contribution.CreatedAt,
        QuantityDelta:   int64(req.Quantity),
    })

Step 3.3: 新增 comboEnqueueParams 与 enqueueComboContribution 函数

在 activity_service.go 文件末尾(最后一个 } 之前)新增:

// comboEnqueueParams 入队聚合贡献所需参数
type comboEnqueueParams struct {
    ActivityID     int64
    UserID         int64
    StarID         int64
    ItemID         int64
    ItemType       string
    ItemName       string
    ItemIcon       string
    Nickname       string
    AvatarURL      string
    FirstID        int64
    FirstCreatedAt int64
    QuantityDelta  int64
}

// enqueueComboContribution 累加聚合 Hash + 仅首次入队 Stream
// - Redis 不可用 → 立即 Publish 一条单 record(降级,保留旧行为)
// - quantity 用 HINCRBY 累加;其他字段 HSetNX 保护首次值不覆盖
// - 入队用 SET NX EX 3 防重复
func (s *activityService) enqueueComboContribution(ctx context.Context, p comboEnqueueParams) {
    if s.redisClient == nil {
        s.publishContributionImmediate(ctx, p)
        return
    }

    aggKey := s.aggKey(p.UserID, p.ItemType)
    lockKey := s.aggLockKey(p.UserID, p.ItemType)

    pipe := s.redisClient.TxPipeline()
    pipe.HIncrBy(ctx, aggKey, "quantity", p.QuantityDelta)
    pipe.HSetNX(ctx, aggKey, "activity_id", p.ActivityID)
    pipe.HSetNX(ctx, aggKey, "user_id", p.UserID)
    pipe.HSetNX(ctx, aggKey, "star_id", p.StarID)
    pipe.HSetNX(ctx, aggKey, "item_id", p.ItemID)
    pipe.HSetNX(ctx, aggKey, "item_name", p.ItemName)
    pipe.HSetNX(ctx, aggKey, "item_icon", p.ItemIcon)
    pipe.HSetNX(ctx, aggKey, "nickname", p.Nickname)
    pipe.HSetNX(ctx, aggKey, "avatar_url", p.AvatarURL)
    pipe.HSetNX(ctx, aggKey, "first_id", p.FirstID)
    pipe.HSetNX(ctx, aggKey, "first_created_at", p.FirstCreatedAt)
    pipe.Expire(ctx, aggKey, comboAggTTL)
    setNX, _ := s.redisClient.SetNX(ctx, lockKey, "1", comboWindowTTL).Result()
    if setNX {
        pipe.XAdd(ctx, &redis.XAddArgs{
            Stream: comboStreamKey,
            MaxLen: comboStreamMaxLen,
            Approx: true,
            Values: map[string]interface{}{
                "activity_id":      p.ActivityID,
                "user_id":          p.UserID,
                "item_type":        p.ItemType,
                "first_id":         p.FirstID,
                "first_created_at": p.FirstCreatedAt,
                "expire_at_ms":     time.Now().Add(comboWindowTTL).UnixMilli(),
            },
        })
    }
    if _, err := pipe.Exec(ctx); err != nil {
        logger.Logger.Warn("combo pipeline failed, fallback to immediate publish", zap.Error(err))
        s.publishContributionImmediate(ctx, p)
    }
}

// publishContributionImmediate 立即 Publish 一条单 record(降级路径)
func (s *activityService) publishContributionImmediate(ctx context.Context, p comboEnqueueParams) {
    if s.redisClient == nil {
        return
    }
    record := &pb.ContributionRecord{
        Id:         p.FirstID,
        UserId:     p.UserID,
        Nickname:   p.Nickname,
        AvatarUrl:  p.AvatarURL,
        StarId:     p.StarID,
        ItemId:     p.ItemID,
        ItemType:   p.ItemType,
        ItemName:   p.ItemName,
        ItemIcon:   p.ItemIcon,
        Quantity:   int32(p.QuantityDelta),
        ComboCount: int32(p.QuantityDelta),
        CreatedAt:  p.FirstCreatedAt,
    }
    payload, _ := json.Marshal(map[string]interface{}{
        "activity_id": p.ActivityID,
        "type":        "contributions_response",
        "record":      record,
    })
    s.redisClient.Publish(ctx, fmt.Sprintf("act:%d:contributions", p.ActivityID), payload)
}

Step 3.4: 验证编译

cd backend && go build ./services/activityService/...

预期: BUILD OK(本任务只动了 PurchaseItem,BatchPurchaseItem / GetLatestContributions 仍有旧调用 — Task 4/5 修复)。

Step 3.5: 写 enqueueComboContribution Redis 不可用降级单测

在 backend/services/activityService/service/activity_service_combo_test.go 新建:

package service

import (
    "context"
    "testing"

    "github.com/stretchr/testify/assert"
)

func TestEnqueueComboContribution_NilRedis_FallbackImmediate(t *testing.T) {
    s := &activityService{
        redisClient:   nil,
        userRPCClient: &mockUserRPC{},
        activityRepo:  &mockActivityRepo{},
    }
    // redisClient 为 nil 时,降级路径不会 Publish(内部判空),直接返回不 panic
    assert.NotPanics(t, func() {
        s.enqueueComboContribution(context.Background(), comboEnqueueParams{
            ActivityID:    1,
            UserID:        100,
            StarID:        7,
            ItemID:        42,
            ItemType:      "firework",
            QuantityDelta: 1,
        })
    })
}

Step 3.6: 运行新增测试

cd backend && go test ./services/activityService/service/ -run TestEnqueueCombo -v

预期: PASS

Step 3.7: 暂不 commit

继续 Task 4。

Task 4: 后端 — 改造 `BatchPurchaseItem` 使用聚合入队

Files:

Modify: backend/services/activityService/service/activity_service.go:742-773
Step 4.1: 定位 BatchPurchaseItem 内的连击 + 推送循环

第 742-773 行是循环内每条 item 的 incrementComboCount + Publish 段。

Step 4.2: 替换循环体

将第 742-773 行整段替换为:

        // 连击聚合:同 user+itemType 窗口内合并推送
        itemName, itemIcon := "", ""
        if activityItem != nil {
            itemName = activityItem.ItemName
            itemIcon = activityItem.IconURL
        }
        s.enqueueComboContribution(ctx, comboEnqueueParams{
            ActivityID:    req.ActivityId,
            UserID:        userID,
            StarID:        req.StarId,
            ItemID:        activityItem.ID,
            ItemType:      item.ItemType,
            ItemName:      itemName,
            ItemIcon:      itemIcon,
            Nickname:      nickname,
            AvatarURL:     avatarURL,
            FirstID:       contribution.ID,
            FirstCreatedAt: contribution.CreatedAt,
            QuantityDelta: int64(item.Quantity),
        })

注意: nickname/avatarURL 已在循环外(第 749 行)计算过,需要把它们的定义移到循环外或保持变量作用域。阅读第 746-749 行确认上下文:它们在循环内 if 块中,作用域只在该 if 块,需上提:

    // 在 for 循环之前
    nickname, avatarURL := "", ""
    if profile, _ := s.userRPCClient.GetFanProfile(userID, req.StarId); profile != nil {
        nickname = profile.Nickname
        avatarURL = profile.AvatarUrl
    }

    // for 循环内删除原 749-751 行 nickname/avatarURL 计算

Step 4.3: 验证编译

cd backend && go build ./services/activityService/...

预期: BUILD OK(GetLatestContributions 仍在调 getComboCount,Task 5 修复)。

Step 4.4: 暂不 commit

继续 Task 5。

Task 5: 后端 — 改造 `GetLatestContributions` 增加内存合并 + 删除 `getComboCount` 引用

Files:

Modify: backend/services/activityService/service/activity_service.go:1266-1307
Step 5.1: 阅读当前组装 records 段

第 1266-1307 行是循环遍历 contributions、组装 *pb.ContributionRecord 的代码。comboCount 在第 1291 行调用 s.getComboCount(ctx, ...),需替换为读取 agg Hash,失败则回退为 1。

Step 5.2: 替换 comboCount 取值

将第 1290-1291 行:

        // 获取连击计数
        comboCount := int32(s.getComboCount(ctx, c.UserID, c.ItemType))

替换为:

        // 聚合计数优先从 agg Hash 读;窗口已过期则回退为 quantity
        comboCount := int32(c.Quantity)
        if s.redisClient != nil {
            if v, err := s.redisClient.HGet(ctx, s.aggKey(c.UserID, c.ItemType), "quantity").Int64(); err == nil && v > 0 {
                comboCount = int32(v)
            }
        }

Step 5.3: 增加 SQL 后内存合并(3 秒窗口按 user_id+item_id 累加)

定位到 records[i] = &pb.ContributionRecord{...} 赋值之后、return 之前(第 1309 行附近),新增:

    // 内存合并:同 (user_id, item_id) 在 3 秒窗口内的行合并为 1 条,
    // quantity 累加,id 取最早行(与 WS first_id 一致,前端按 id 去重不会双计)
    const COMBO_WINDOW_MS int64 = 3000
    merged := make([]*pb.ContributionRecord, 0, len(records))
    i := 0
    for i < len(records) {
        first := records[i]
        var totalQty int32 = first.Quantity
        j := i + 1
        for j < len(records) &&
            records[j].UserId == first.UserId &&
            records[j].ItemId == first.ItemId &&
            records[j].CreatedAt-first.CreatedAt <= COMBO_WINDOW_MS {
            totalQty += records[j].Quantity
            j++
        }
        merged = append(merged, &pb.ContributionRecord{
            Id:         first.Id,
            UserId:     first.UserId,
            Nickname:   first.Nickname,
            AvatarUrl:  first.AvatarUrl,
            StarId:     first.StarId,
            ItemId:     first.ItemId,
            ItemType:   first.ItemType,
            ItemName:   first.ItemName,
            ItemIcon:   first.ItemIcon,
            Quantity:   totalQty,
            ComboCount: totalQty,
            CreatedAt:  first.CreatedAt,
        })
        i = j
    }
    records = merged

Step 5.4: 修改 return 处的 records 引用

第 1309-1315 行 return &pb.GetLatestContributionsResponse{...Records: records...} 应保留为 Records: records(已经是引用,无需改)。

Step 5.5: 验证编译

cd backend && go build ./services/activityService/...

预期: BUILD OK(无旧函数引用残留)。

Step 5.6: 写内存合并单测

追加到 backend/services/activityService/service/activity_service_combo_test.go:

func TestGetLatestContributions_MergesComboRecords(t *testing.T) {
    // 直接验证合并逻辑:GetLatestContributions 内的内存合并
    // SQL 已按 created_at DESC 返回行,合并按 3 秒窗口 + (user,item) 分组
    records := []*pb.ContributionRecord{
        {Id: 10, UserId: 100, ItemId: 1, ItemType: "x", Quantity: 1, CreatedAt: 1700000003000},
        {Id: 11, UserId: 100, ItemId: 1, ItemType: "x", Quantity: 1, CreatedAt: 1700000002000},
        {Id: 12, UserId: 100, ItemId: 1, ItemType: "x", Quantity: 1, CreatedAt: 1700000001000},
        {Id: 13, UserId: 100, ItemId: 2, ItemType: "y", Quantity: 1, CreatedAt: 1700000000500},
    }

    const COMBO_WINDOW_MS int64 = 3000
    merged := make([]*pb.ContributionRecord, 0, len(records))
    i := 0
    for i < len(records) {
        first := records[i]
        totalQty := first.Quantity
        j := i + 1
        for j < len(records) &&
            records[j].UserId == first.UserId &&
            records[j].ItemId == first.ItemId &&
            records[j].CreatedAt-first.CreatedAt <= COMBO_WINDOW_MS {
            totalQty += records[j].Quantity
            j++
        }
        merged = append(merged, &pb.ContributionRecord{
            Id: first.Id, Quantity: totalQty, ComboCount: totalQty,
        })
        i = j
    }

    assert.Len(t, merged, 2, "3 条同 (user,item) 合并 + 1 条不同 item")
    assert.Equal(t, int32(3), merged[0].Quantity)
    assert.Equal(t, int64(12), merged[0].Id, "合并 id 取最早(同组最小 id)")
    assert.Equal(t, int32(1), merged[1].Quantity)
}

func TestGetLatestContributions_ComboWindowBoundary(t *testing.T) {
    // 跨 3 秒窗口不合并
    records := []*pb.ContributionRecord{
        {Id: 20, UserId: 100, ItemId: 1, Quantity: 1, CreatedAt: 5000},
        {Id: 21, UserId: 100, ItemId: 1, Quantity: 1, CreatedAt: 1000}, // 差 4000ms > 3000
    }
    const COMBO_WINDOW_MS int64 = 3000
    merged := make([]*pb.ContributionRecord, 0, len(records))
    i := 0
    for i < len(records) {
        first := records[i]
        totalQty := first.Quantity
        j := i + 1
        for j < len(records) &&
            records[j].UserId == first.UserId &&
            records[j].ItemId == first.ItemId &&
            records[j].CreatedAt-first.CreatedAt <= COMBO_WINDOW_MS {
            totalQty += records[j].Quantity
            j++
        }
        merged = append(merged, &pb.ContributionRecord{Id: first.Id, Quantity: totalQty})
        i = j
    }
    assert.Len(t, merged, 2, "跨窗口不合并")
}

Step 5.7: 运行合并测试

cd backend && go test ./services/activityService/service/ -run "TestGetLatestContributions_MergesComboRecords|TestGetLatestContributions_ComboWindowBoundary" -v

预期: 2 tests passed

Step 5.8: 运行整套 service 包测试,确认未破坏既有测试

cd backend && go test ./services/activityService/service/ -v

预期: 全部通过(包括 TestBuildComboRecord_* / TestEnqueueComboContribution_* / 既有 TestGetTop* 等)

Step 5.9: 暂不 commit

继续 Task 6。

Task 6: 后端 — 在 `main.go` 启动 combo worker

Files:

Modify: backend/services/activityService/main.go:59-122
Step 6.1: 阅读 main() 函数

第 59-122 行,定位 activityService := service.NewActivityService(...) 那一行。

Step 6.2: 在 service 初始化之后启动 worker

在 activityService := service.NewActivityService(...) 之后、activityProvider := provider.NewActivityProvider(activityService) 之前新增:

	// 启动 combo Stream worker(3 秒连击合并推送)
	workerCtx, workerCancel := context.WithCancel(context.Background())
	defer workerCancel()
	activityService.StartComboStreamWorker(workerCtx)

Step 6.3: 验证编译

cd backend && go build ./services/activityService/...

预期: BUILD OK

Step 6.4: 暂不 commit

继续 Task 7。

Task 7: 前端 — 新增 `mergeComboRecords` 工具函数

Files:

Modify: frontend/pages/support-activity/composables/useContributionRealtime.js
Step 7.1: 在文件顶部新增合并常量与函数

在 useContributionRealtime.js 的 import 之后(第 4 行 import 之后,第 5 行注释之前)插入:

/**
 * 连击合并常量与工具函数 —— WS 与轮询两条路径共用
 * - 同一 user_id+item_id 在 COMBO_WINDOW_MS 内的多条 → 合并为 1 条
 * - 合并后 id = 同组最早 id(与后端 first_id 一致,前端按 id 去重不会双计)
 * - quantity = 同组 quantity 之和
 */
const COMBO_WINDOW_MS = 3000

export function mergeComboRecords(records) {
    if (!Array.isArray(records) || records.length === 0) return records

    // 倒序遍历:新→旧;同 key 在窗口内合并到索引更大的(更新的)那条
    const result = [...records]
    const indexByKey = new Map() // key = `${user_id}:${item_id}` -> result 中索引

    for (let i = result.length - 1; i >= 0; i--) {
        const r = result[i]
        const key = `${r.user_id}:${r.item_id}`
        const existingIdx = indexByKey.get(key)
        if (existingIdx !== undefined) {
            const existing = result[existingIdx]
            if (Math.abs(existing.created_at - r.created_at) <= COMBO_WINDOW_MS) {
                existing.quantity += r.quantity
                existing.combo_count = existing.quantity
                if (r.id < existing.id) existing.id = r.id // 取最早 id
                result.splice(i, 1)
                continue
            }
        }
        indexByKey.set(key, i)
    }
    return result
}

Step 7.2: 在 onWsMessage 中调用合并

第 33 行 records.value = [...records.value, record].slice(-MAX_RECORDS) 替换为:

            records.value = mergeComboRecords([...records.value, record]).slice(-MAX_RECORDS)

Step 7.3: 暂不 commit

继续 Task 8。

Task 8: 前端 — `useContributionPolling.js` 在轮询路径调用合并

Files:

Modify: frontend/pages/support-activity/composables/useContributionPolling.js
Step 8.1: 引入 mergeComboRecords

在第 2 行 import { getActivityContributionsLatestApi } from '@/utils/api.js' 之后新增:

import { mergeComboRecords } from './useContributionRealtime.js'

Step 8.2: 在 fetchLatest 合并新记录后再调用合并

定位到第 86 行 records.value = [...newRecords, ...records.value].slice(0, MAX_RECORDS),替换为:

        records.value = mergeComboRecords([...newRecords, ...records.value]).slice(0, MAX_RECORDS)

Step 8.3: 暂不 commit

继续 Task 9。

Task 9: 前端 — 核对并按需调整 `ContributionList.vue` 模板

Files:

Modify: frontend/pages/support-activity/components/ContributionList.vue (仅当模板仍引用 combo_count 时)
Step 9.1: 核对当前模板

读取 ContributionList.vue 第 40-46 行。当前现状(核对后)已使用 record.quantity:

<text class="item-x">X</text>
<text
  class="item-count"
  :class="getCountSizeClass(record.quantity)"
  >{{ record.quantity }}</text>

若已如上:跳至 Step 9.3。

若仍引用 combo_count:把 :class="getCountSizeClass(record.combo_count > 1 ? record.combo_count : record.quantity)" 与内层 {{ record.combo_count > 1 ? record.combo_count : record.quantity }} 改为 record.quantity。

Step 9.2: 若模板已正确,无需修改

直接在 commit 时记一笔:template already uses record.quantity, spec §5.4 already satisfied。

Step 9.3: 暂不 commit

继续 Task 10。

Task 10: 集成验证与回归检查

Files: (无代码改动,只跑命令)

Step 10.1: 后端全量测试

cd backend && go test ./services/activityService/... -v

预期: 全部通过。

Step 10.2: 前端 lint(如可用)

cd frontend && npm run lint 2>/dev/null || echo "no lint configured"

预期: 无错误。

Step 10.3: 后端构建 release binary 确认无编译问题

cd backend && go build -o /tmp/activityService ./services/activityService/

预期: BUILD OK。

Step 10.4: 核对接口签名一致

grep -n "ComboCount\|combo_count" \
  backend/services/activityService/service/activity_service.go \
  backend/pkg/proto/activity/*.go 2>/dev/null

预期: 字段保留(用于前端展示兼容 / proto 兼容);但 activity_service.go 中已不再调 getComboCount/incrementComboCount。

Step 10.5: 手动集成测试(如本地 Redis 可用)

# 1. 启动 activityService(假设本地 Redis 已起)
# 2. 用 Postman/curl 调 PurchaseItem 3 次同 user 同 item,间隔 < 3s
# 3. 观察 Redis Stream combo:stream:contributions 中只有 1 条 XAdd
# 4. 等 3 秒后,Redis Channel act:{activityId}:contributions 收到 1 条合并 record(quantity=3)

Task 11: 自审 — 修复后回归检查(spec §8 + CLAUDE.md「自审与回归检查规范」)

Step 11.1: 核对改动文件清单

git status

预期包含:

M backend/services/activityService/service/activity_service.go
A backend/services/activityService/service/combo_worker.go
A backend/services/activityService/service/combo_worker_test.go
A backend/services/activityService/service/activity_service_combo_test.go
M backend/services/activityService/main.go
M frontend/pages/support-activity/composables/useContributionRealtime.js
M frontend/pages/support-activity/composables/useContributionPolling.js
(可能) M frontend/pages/support-activity/components/ContributionList.vue
Step 11.2: 核对下游不受影响

通过 mcp__code-review-graph__query_graph_tool pattern="callers_of" target="comboKey" 检查是否还有外部调用被破坏。预期: 仅同文件内调用。

Step 11.3: 核对 DB schema 未改动

git diff --stat HEAD~0 -- backend/services/activityService/repository/ backend/pkg/models/

预期: 无改动。

Step 11.4: 核对 combo_count 外部消费方

grep -rn "combo_count\|ComboCount" frontend/ --include="*.vue" --include="*.js"

预期: 仅 ContributionList.vue / useContributionRealtime.js / useContributionPolling.js / 后端 proto 文件,无其他业务方。

Step 11.5: 跑一遍 git diff --stat 确认无意外文件

git diff --stat

预期: 改动文件清单与 Step 11.1 完全一致。

Task 12: 提交(由用户在确认后执行;AI 不得主动 commit)

Files: (无代码改动)

Step 12.1: 把改动文件暂存(待用户确认)

git add \
  backend/services/activityService/service/activity_service.go \
  backend/services/activityService/service/combo_worker.go \
  backend/services/activityService/service/combo_worker_test.go \
  backend/services/activityService/service/activity_service_combo_test.go \
  backend/services/activityService/main.go \
  frontend/pages/support-activity/composables/useContributionRealtime.js \
  frontend/pages/support-activity/composables/useContributionPolling.js

Step 12.2: 输出待 commit 信息给用户确认

向用户输出建议的 commit message(按 CLAUDE.md 规范):

feat(contribution):连击贡献合并推送(Redis Stream 延迟队列)

- PurchaseItem/BatchPurchaseItem 改用 agg Hash + Stream 延迟队列,
  3 秒窗口内同 (user, item) 多次购买合并为 1 条 WS 推送
- 新增 combo_worker goroutine 从 Stream 取出到期条目 → 读 agg Hash → Publish 合并 record
- GetLatestContributions 在 SQL 后做 3 秒窗口内存合并
- 前端 useContributionRealtime/Polling 两条路径共用 mergeComboRecords 工具函数

可靠性:
- Stream + Consumer Group 重启不丢推送(pending list 自动重投递)
- Redis 故障降级为立即 Publish(与旧行为一致)

Refs: docs/superpowers/specs/2026-06-23-contribution-combo-aggregation-design.md

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

注:按 CLAUDE.md「AI 不得主动 commit」规范,此步骤由用户确认后手动执行 commit;AI 不运行 git commit。

风险与回滚

风险	回滚方案
Stream 写入失败导致推送丢失	enqueueComboContribution 内已加 `pipe.Exec` 失败 fallback 到 `publishContributionImmediate`
worker 启动失败	`s.redisClient == nil` 时直接 warn return;main.go 不因此崩溃
内存合并逻辑与 WS 推送 ID 不一致	合并后 `Id = first.Id`,与 WS 推送的 `first_id` 一致
前端模板未及时改	既有模板已用 `quantity`(Task 9 核对后确认),无遗漏分支
多副本场景下 XAdd 重复入队	SET NX EX 3 防重复;Consumer Group + XAUTOCLAIM 保证推送幂等

监控指标(可选,部署后配置)

combo:stream:contributions Stream 长度(应 < 1000,超过说明 worker 处理慢)
XPENDING combo:stream:contributions combo-publishers pending 数量(应 < 100)
前端 mergeComboRecords 调用次数与合并前后 record 数差值

Self-Review Checklist

Spec §1-§4 覆盖:Task 1-6
Spec §5 前端:Task 7-9
Spec §6 错误处理:Task 3/5 已覆盖 Redis 不可用降级、processComboEntry 空 hash skip
Spec §7 性能:Stream MAXLEN 100000 + Pipeline(TxPipeline)减少 RTT
Spec §8 影响范围:Task 11 自审核对
Spec §9 测试:Task 2/3/5/10 含单测与集成验证
Spec §10 风险:Task 11 自审 + 风险表覆盖
Spec §11 实施步骤 1-9:对应 Task 1-9
类型/函数名一致性:aggKey / aggLockKey / enqueueComboContribution / publishContributionImmediate / buildComboRecord / StartComboStreamWorker / processComboEntry / mergeComboRecords 在所有 Task 中用法一致
无 placeholder:每个代码块都是完整可直接复制的代码

37 KiB Raw Blame History Unescape Escape

用户贡献连击聚合推送 — 实施计划

File Structure

后端 (Go)

前端 (Vue / uni-app)

测试

任务清单

Task 1: 后端 — 删除旧的 incrementComboCount 与 getComboCount,新增聚合键生成函数

Task 2: 后端 — 新增 combo_worker.go(Stream 消费者 + 聚合辅助函数)

Task 3: 后端 — 改造 PurchaseItem 使用 agg Hash + XAdd,Redis 不可用时降级

Task 4: 后端 — 改造 BatchPurchaseItem 使用聚合入队

Task 5: 后端 — 改造 GetLatestContributions 增加内存合并 + 删除 getComboCount 引用

Task 6: 后端 — 在 main.go 启动 combo worker

Task 7: 前端 — 新增 mergeComboRecords 工具函数

Task 8: 前端 — useContributionPolling.js 在轮询路径调用合并

Task 9: 前端 — 核对并按需调整 ContributionList.vue 模板

Task 10: 集成验证与回归检查

Task 11: 自审 — 修复后回归检查(spec §8 + CLAUDE.md「自审与回归检查规范」)

Task 12: 提交(由用户在确认后执行;AI 不得主动 commit)

风险与回滚

监控指标(可选,部署后配置)

Self-Review Checklist

37 KiB

Raw Blame History

Task 1: 后端 — 删除旧的 `incrementComboCount` 与 `getComboCount`,新增聚合键生成函数

Task 2: 后端 — 新增 `combo_worker.go`(Stream 消费者 + 聚合辅助函数)

Task 3: 后端 — 改造 `PurchaseItem` 使用 agg Hash + XAdd,Redis 不可用时降级

Task 4: 后端 — 改造 `BatchPurchaseItem` 使用聚合入队

Task 5: 后端 — 改造 `GetLatestContributions` 增加内存合并 + 删除 `getComboCount` 引用

Task 6: 后端 — 在 `main.go` 启动 combo worker

Task 7: 前端 — 新增 `mergeComboRecords` 工具函数

Task 8: 前端 — `useContributionPolling.js` 在轮询路径调用合并

Task 9: 前端 — 核对并按需调整 `ContributionList.vue` 模板