xiaoyu/topfans
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat:增加实时留言功能和修改轮询道具购买记录显示改为websocket显示

Browse Source
This commit is contained in:
zerosaturation 2026-06-23 18:13:55 +08:00
parent c6f2a86467
commit 75ab617c6c
31 changed files with 4657 additions and 164 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

6
backend/gateway/config/config.go
View File

@ -122,7 +122,8 @@ func (c *OSSConfig) GetUploadDir(uploadType string) string {
// WebSocketConfig WebSocket 配置
type WebSocketConfig struct {
AIChatPath string // WebSocket 路径，默认 /ai-chat
AIChatPath string // AI Chat WebSocket 路径，默认 /ai-chat
ActivityPath string // 活动实时推送 WS 路径，默认 /activity
}
// Load 加载配置
@ -193,7 +194,8 @@ func Load() *Config {
TimeZone: getEnv("DB_TIMEZONE", "Asia/Shanghai"),
},
WebSocket: WebSocketConfig{
AIChatPath: getEnv("WS_AI_CHAT_PATH", "/ai-chat"),
AIChatPath: getEnv("WS_AI_CHAT_PATH", "/ai-chat"),
ActivityPath: getEnv("WS_ACTIVITY_PATH", "/activity"),
},
}
}

182
backend/gateway/controller/activity_controller.go
View File

@ -730,6 +730,188 @@ func convertLatestContributionsResponse(resp *pbActivity.GetLatestContributionsR
}
}
// ListActivityMessages 列出活动留言
// @Summary 列出活动留言
// @Description 分页获取活动留言列表（最新在上）
// @Tags activities
// @Accept json
// @Produce json
// @Security BearerAuth
// @Param activity_id path int64 true "活动ID"
// @Param page query int false "页码，默认1"
// @Param page_size query int false "每页数量，默认20，最大50"
// @Success 200 {object} response.Response
// @Router /api/v1/activities/{activity_id}/messages [get]
func (ctrl *ActivityController) ListActivityMessages(c *gin.Context) {
// 解析路径参数
activityIDStr := c.Param("id")
activityID, err := strconv.ParseInt(activityIDStr, 10, 64)
if err != nil {
response.Error(c, http.StatusBadRequest, "活动ID参数错误")
return
}
// 解析查询参数
page, _ := strconv.Atoi(c.DefaultQuery("page", "1"))
pageSize, _ := strconv.Atoi(c.DefaultQuery("page_size", "20"))
if page <= 0 {
page = 1
}
if pageSize <= 0 {
pageSize = 20
}
if pageSize > 50 {
pageSize = 50
}
logger.Logger.Info("ListActivityMessages request",
zap.Int64("activity_id", activityID),
zap.Int("page", page),
zap.Int("page_size", pageSize),
)
// 设置上下文
ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
defer cancel()
// 调用 RPC
resp, err := ctrl.activityService.ListActivityMessages(ctx, &pbActivity.ListActivityMessagesRequest{
ActivityId: activityID,
Page: int32(page),
PageSize: int32(pageSize),
})
if err != nil {
logger.Logger.Error("ListActivityMessages RPC failed", zap.Error(err))
response.Error(c, http.StatusInternalServerError, "获取活动留言失败")
return
}
if resp.Base.Code != uint32(codes.OK) {
response.ErrorWithCode(c, int(resp.Base.Code), resp.Base.Message)
return
}
// 转换响应
messages := make([]map[string]interface{}, 0, len(resp.Messages))
for _, m := range resp.Messages {
messages = append(messages, map[string]interface{}{
"id": m.Id,
"activity_id": m.ActivityId,
"user_id": m.UserId,
"star_id": m.StarId,
"nickname": m.Nickname,
"avatar_url": m.AvatarUrl,
"content": m.Content,
"created_at": m.CreatedAt,
})
}
response.Success(c, map[string]interface{}{
"messages": messages,
"page": resp.Page,
"page_size": resp.PageSize,
"total": resp.Total,
})
}
// CreateActivityMessage 发送一条活动留言
// @Summary 发送活动留言
// @Description 用户在应援活动页面发送一条祝福留言
// @Tags activities
// @Accept json
// @Produce json
// @Security BearerAuth
// @Param activity_id path int64 true "活动ID"
// @Param request body object{content=string} true "留言内容"
// @Success 200 {object} response.Response
// @Router /api/v1/activities/{activity_id}/messages [post]
func (ctrl *ActivityController) CreateActivityMessage(c *gin.Context) {
// 从上下文获取用户信息
userID, exists := c.Get("user_id")
if !exists {
response.Error(c, http.StatusUnauthorized, "未授权")
return
}
starID, exists := c.Get("star_id")
if !exists {
response.Error(c, http.StatusUnauthorized, "未授权")
return
}
// 解析路径参数
activityIDStr := c.Param("id")
activityID, err := strconv.ParseInt(activityIDStr, 10, 64)
if err != nil {
response.Error(c, http.StatusBadRequest, "活动ID参数错误")
return
}
// 解析请求体
var req struct {
Content string `json:"content" binding:"required"`
}
if err := c.ShouldBindJSON(&req); err != nil {
response.Error(c, http.StatusBadRequest, "请求参数错误")
return
}
if req.Content == "" {
response.Error(c, http.StatusBadRequest, "content 是必填参数")
return
}
logger.Logger.Info("CreateActivityMessage request",
zap.Int64("user_id", userID.(int64)),
zap.Int64("star_id", starID.(int64)),
zap.Int64("activity_id", activityID),
zap.Int("content_len", len(req.Content)),
)
// 设置上下文（带 attachments 传递 user_id / star_id 给下游）
ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
defer cancel()
ctx = context.WithValue(ctx, constant.AttachmentKey, map[string]interface{}{
"user_id": strconv.FormatInt(userID.(int64), 10),
"star_id": strconv.FormatInt(starID.(int64), 10),
})
// 调用 RPC
resp, err := ctrl.activityService.CreateActivityMessage(ctx, &pbActivity.CreateActivityMessageRequest{
ActivityId: activityID,
UserId: userID.(int64),
StarId: starID.(int64),
Content: req.Content,
})
if err != nil {
logger.Logger.Error("CreateActivityMessage RPC failed", zap.Error(err))
response.Error(c, http.StatusInternalServerError, "发送活动留言失败")
return
}
if resp.Base.Code != uint32(codes.OK) {
response.ErrorWithCode(c, int(resp.Base.Code), resp.Base.Message)
return
}
// 返回新创建的留言
m := resp.Message
response.Success(c, map[string]interface{}{
"message": map[string]interface{}{
"id": m.Id,
"activity_id": m.ActivityId,
"user_id": m.UserId,
"star_id": m.StarId,
"nickname": m.Nickname,
"avatar_url": m.AvatarUrl,
"content": m.Content,
"created_at": m.CreatedAt,
},
})
}
// convertMintingActivitiesResponse 转换铸造活动列表响应
func convertMintingActivitiesResponse(resp *pbActivity.GetMintingActivitiesResponse) map[string]interface{} {
activities := make([]map[string]interface{}, 0, len(resp.Activities))

14
backend/gateway/main.go
View File

@ -1,6 +1,7 @@
package main
import (
"context"
"fmt"
"os"
"os/signal"
@ -18,6 +19,7 @@ import (
"go.uber.org/zap"
docs "github.com/topfans/backend/gateway/docs"
"github.com/topfans/backend/gateway/socket"
pbModeration "github.com/topfans/backend/pkg/proto/moderation"
)
@ -220,9 +222,19 @@ func main() {
logger.Logger.Fatal("Failed to create ModerationService pb client", zap.Error(err))
}
// 4.13 初始化 Activity Hub（WebSocket 实时推送）
redisClient := database.GetRedis()
activityHub := socket.NewActivityHub(redisClient, cfg.WebSocket.ActivityPath)
go activityHub.Run(context.Background())
defer activityHub.Close()
logger.Logger.Info("ActivityHub initialized",
zap.String("path", cfg.WebSocket.ActivityPath),
zap.Bool("redis_available", redisClient != nil),
)
// 5. 设置路由
logger.Logger.Info("Setting up routes...")
r, err := router.SetupRouter(userClient, socialClient, assetClient, galleryClient, activityClient, taskClient, starbookClient, aiChatClient, statisticClient, notificationClient, modSvc, cfg.WebSocket.AIChatPath)
r, err := router.SetupRouter(userClient, socialClient, assetClient, galleryClient, activityClient, taskClient, starbookClient, aiChatClient, statisticClient, notificationClient, modSvc, cfg.WebSocket.AIChatPath, activityHub)
if err != nil {
logger.Logger.Fatal("Failed to setup router", zap.Error(err))
}

7
backend/gateway/pkg/response/response.go
View File

@ -146,6 +146,13 @@ func CleanErrorMessage(err error) string {
"user info not found in Dubbo attachments": "请先登录",
"missing or invalid authorization token": "Token缺失或无效",
"已达到最大好友数量限制": "已达到最大好友数量限制",
"活动留言不存在": "活动留言不存在",
"留言太频繁": "留言太频繁，请稍后再试",
"当前活动留言已达上限": "当前活动留言已达上限",
"留言内容不能为空": "留言内容不能为空",
"留言内容过长": "留言内容过长，最多500字",
"留言内容包含不当内容": "留言内容包含不当内容，请修改",
"活动不在进行中": "活动未开始或已结束",
}
msgLower := strings.ToLower(msg)

9
backend/gateway/router/router.go
View File

@ -16,7 +16,7 @@ import (
)
// SetupRouter 设置路由
func SetupRouter(userClient *client.Client, socialClient *client.Client, assetClient *client.Client, galleryClient *client.Client, activityClient *client.Client, taskClient *client.Client, starbookClient *client.Client, aiChatClient *client.Client, statisticClient *client.Client, notificationClient *client.Client, modSvc pbModeration.ModerationService, aiChatPath string) (*gin.Engine, error) {
func SetupRouter(userClient *client.Client, socialClient *client.Client, assetClient *client.Client, galleryClient *client.Client, activityClient *client.Client, taskClient *client.Client, starbookClient *client.Client, aiChatClient *client.Client, statisticClient *client.Client, notificationClient *client.Client, modSvc pbModeration.ModerationService, aiChatPath string, activityHub *socket.ActivityHub) (*gin.Engine, error) {
r := gin.Default()
// 全局中间件
@ -46,6 +46,11 @@ func SetupRouter(userClient *client.Client, socialClient *client.Client, assetCl
aiChatHub.HandleWebSocket(w, r)
}))
// Activity 实时推送 WebSocket 路由
r.GET(activityHub.ActivityPath(), gin.WrapF(func(w http.ResponseWriter, r *http.Request) {
activityHub.HandleWebSocket(w, r)
}))
// 创建控制器
authCtrl, err := controller.NewAuthController(userClient)
if err != nil {
@ -381,6 +386,8 @@ func SetupRouter(userClient *client.Client, socialClient *client.Client, assetCl
activities.POST("/:id/batch-purchase", activityCtrl.BatchPurchaseItem) // 批量购买道具
activities.GET("/:id/ranking", activityCtrl.GetContributionRanking) // 获取贡献点排名
activities.GET("/:id/contributions/latest", activityCtrl.GetLatestContributions) // 获取最新贡献记录
activities.GET("/:id/messages", activityCtrl.ListActivityMessages) // 获取活动留言列表
activities.POST("/:id/messages", activityCtrl.CreateActivityMessage) // 发送一条活动留言
}
// 铸造活动相关路由（运营banner）- 公开接口，不需要认证

368
backend/gateway/socket/activity_socket.go Normal file
View File

@ -0,0 +1,368 @@
package socket
import (
"context"
"encoding/json"
"fmt"
"net/http"
"strconv"
"strings"
"sync"
"time"
"github.com/gorilla/websocket"
"github.com/redis/go-redis/v9"
"github.com/topfans/backend/pkg/jwt"
"github.com/topfans/backend/pkg/logger"
"go.uber.org/zap"
)
// ActivityHub 管理所有 Activity WebSocket 连接
type ActivityHub struct {
clients map[int64]map[*ActivityConn]struct{} // userId -> set of conns
subscriptions map[string]map[*ActivityConn]struct{} // "act:42:messages" / "act:42:contributions" -> conns
redisClient *redis.Client
activityPath string
mu sync.RWMutex
}
// ActivityConn 单条 WebSocket 连接
type ActivityConn struct {
UserID int64
StarID int64
Conn *websocket.Conn
Send chan []byte
Hub *ActivityHub
writeMu sync.Mutex
}
// writeJSON 线程安全 JSON 写入
func (c *ActivityConn) writeJSON(data interface{}) error {
c.writeMu.Lock()
defer c.writeMu.Unlock()
return c.Conn.WriteJSON(data)
}
// NewActivityHub 创建 ActivityHub
func NewActivityHub(redisClient *redis.Client, activityPath string) *ActivityHub {
return &ActivityHub{
clients: make(map[int64]map[*ActivityConn]struct{}),
subscriptions: make(map[string]map[*ActivityConn]struct{}),
redisClient: redisClient,
activityPath: activityPath,
}
}
// ActivityPath 返回配置的 WebSocket 路径
func (h *ActivityHub) ActivityPath() string {
return h.activityPath
}
// Run 启动 Redis PSubscribe，收到 publish 后 fanout 到本地连接
func (h *ActivityHub) Run(ctx context.Context) {
if h.redisClient == nil {
logger.Logger.Warn("ActivityHub: redisClient is nil, Pub/Sub fanout disabled")
<-ctx.Done()
return
}
sub := h.redisClient.PSubscribe(ctx, "act:*:messages", "act:*:contributions")
defer sub.Close()
ch := sub.Channel()
logger.Logger.Info("ActivityHub subscribed to Redis Pub/Sub channels")
for {
select {
case <-ctx.Done():
logger.Logger.Info("ActivityHub Run loop exiting due to context done")
return
case msg, ok := <-ch:
if !ok {
logger.Logger.Warn("ActivityHub Redis Pub/Sub channel closed")
return
}
var payload map[string]interface{}
if err := json.Unmarshal([]byte(msg.Payload), &payload); err != nil {
logger.Logger.Error("ActivityHub failed to unmarshal pubsub payload", zap.Error(err))
continue
}
h.fanout(msg.Channel, payload)
}
}
}
// fanout 把 payload 推送到订阅该 channel 的所有本地连接
func (h *ActivityHub) fanout(channel string, payload map[string]interface{}) {
h.mu.RLock()
conns := h.subscriptions[channel]
targets := make([]*ActivityConn, 0, len(conns))
for c := range conns {
targets = append(targets, c)
}
h.mu.RUnlock()
for _, c := range targets {
if err := c.writeJSON(payload); err != nil {
logger.Logger.Error("ActivityHub writeJSON failed", zap.Int64("user_id", c.UserID), zap.Error(err))
}
}
}
// HandleWebSocket 处理 /activity 握手
func (h *ActivityHub) HandleWebSocket(w http.ResponseWriter, r *http.Request) {
token := r.URL.Query().Get("token")
userID, starID, err := h.validateToken(token)
if err != nil {
logger.Logger.Error("Activity WebSocket token validation failed", zap.Error(err))
w.WriteHeader(http.StatusUnauthorized)
_ = json.NewEncoder(w).Encode(map[string]interface{}{
"type": "auth_response",
"success": false,
"error": "invalid_token",
})
return
}
conn, err := upgrader.Upgrade(w, r, nil)
if err != nil {
logger.Logger.Error("Activity WebSocket upgrade failed", zap.Error(err))
return
}
c := &ActivityConn{
UserID: userID,
StarID: starID,
Conn: conn,
Send: make(chan []byte, 256),
Hub: h,
}
h.mu.Lock()
if h.clients[userID] == nil {
h.clients[userID] = make(map[*ActivityConn]struct{})
}
h.clients[userID][c] = struct{}{}
h.mu.Unlock()
logger.Logger.Info("Activity WebSocket connection established",
zap.Int64("user_id", userID),
zap.Int64("star_id", starID),
)
// 立即推送 auth_response
_ = conn.WriteJSON(map[string]interface{}{
"type": "auth_response",
"success": true,
"user_id": userID,
"star_id": starID,
})
go c.readPump()
go c.writePump()
}
// validateToken 验证 token（JWT）
func (h *ActivityHub) validateToken(token string) (int64, int64, error) {
if strings.HasPrefix(token, "Bearer_") {
token = strings.TrimPrefix(token, "Bearer_")
}
claims, err := jwt.ParseToken(token)
if err != nil {
return 0, 0, fmt.Errorf("failed to parse token: %w", err)
}
if claims.UserID == 0 {
return 0, 0, fmt.Errorf("invalid user id")
}
return claims.UserID, claims.StarID, nil
}
// readPump 读取客户端消息
func (c *ActivityConn) readPump() {
defer func() {
c.Hub.unregister(c)
c.Conn.Close()
}()
c.Conn.SetReadLimit(64 * 1024) // 64KB
c.Conn.SetReadDeadline(time.Now().Add(60 * time.Second))
c.Conn.SetPongHandler(func(string) error {
c.Conn.SetReadDeadline(time.Now().Add(60 * time.Second))
return nil
})
for {
_, message, err := c.Conn.ReadMessage()
if err != nil {
if websocket.IsUnexpectedCloseError(err, websocket.CloseGoingAway, websocket.CloseAbnormalClosure) {
logger.Logger.Error("Activity WebSocket read error", zap.Error(err))
}
break
}
var msg map[string]interface{}
if err := json.Unmarshal(message, &msg); err != nil {
logger.Logger.Error("Failed to parse activity message", zap.Error(err))
continue
}
c.handleMessage(msg)
}
}
// writePump 写消息到客户端
func (c *ActivityConn) writePump() {
ticker := time.NewTicker(30 * time.Second)
defer func() {
ticker.Stop()
c.Conn.Close()
}()
for {
select {
case message, ok := <-c.Send:
c.Conn.SetWriteDeadline(time.Now().Add(10 * time.Second))
if !ok {
_ = c.Conn.WriteMessage(websocket.CloseMessage, []byte{})
return
}
w, err := c.Conn.NextWriter(websocket.TextMessage)
if err != nil {
return
}
_, _ = w.Write(message)
if err := w.Close(); err != nil {
return
}
case <-ticker.C:
c.Conn.SetWriteDeadline(time.Now().Add(10 * time.Second))
if err := c.Conn.WriteMessage(websocket.PingMessage, nil); err != nil {
return
}
}
}
}
// handleMessage 处理客户端 subscribe/unsubscribe/ping
func (c *ActivityConn) handleMessage(msg map[string]interface{}) {
action, _ := msg["action"].(string)
switch action {
case "ping":
c.Send <- []byte(`{"type":"pong"}`)
case "subscribe":
activityID := toInt64(msg["activity_id"])
topics := toStringSlice(msg["topics"])
c.Hub.subscribe(c, activityID, topics)
_ = c.writeJSON(map[string]interface{}{
"type": "subscribe_response",
"activity_id": activityID,
"topics": topics,
})
case "unsubscribe":
activityID := toInt64(msg["activity_id"])
topics := toStringSlice(msg["topics"])
c.Hub.unsubscribe(c, activityID, topics)
_ = c.writeJSON(map[string]interface{}{
"type": "unsubscribe_response",
"activity_id": activityID,
"topics": topics,
})
default:
logger.Logger.Warn("Unknown activity action", zap.String("action", action))
}
}
// subscribe 幂等订阅
func (h *ActivityHub) subscribe(c *ActivityConn, activityID int64, topics []string) {
if activityID <= 0 || len(topics) == 0 {
return
}
h.mu.Lock()
defer h.mu.Unlock()
for _, t := range topics {
ch := fmt.Sprintf("act:%d:%s", activityID, t)
if h.subscriptions[ch] == nil {
h.subscriptions[ch] = make(map[*ActivityConn]struct{})
}
h.subscriptions[ch][c] = struct{}{}
}
}
// unsubscribe 幂等取消订阅
func (h *ActivityHub) unsubscribe(c *ActivityConn, activityID int64, topics []string) {
if activityID <= 0 || len(topics) == 0 {
return
}
h.mu.Lock()
defer h.mu.Unlock()
for _, t := range topics {
ch := fmt.Sprintf("act:%d:%s", activityID, t)
if conns, ok := h.subscriptions[ch]; ok {
delete(conns, c)
if len(conns) == 0 {
delete(h.subscriptions, ch)
}
}
}
}
// unregister 断开时清理
func (h *ActivityHub) unregister(c *ActivityConn) {
h.mu.Lock()
defer h.mu.Unlock()
if conns, ok := h.clients[c.UserID]; ok {
delete(conns, c)
if len(conns) == 0 {
delete(h.clients, c.UserID)
}
}
for ch, conns := range h.subscriptions {
if _, ok := conns[c]; ok {
delete(conns, c)
if len(conns) == 0 {
delete(h.subscriptions, ch)
}
}
}
}
// Close 关闭所有连接
func (h *ActivityHub) Close() {
h.mu.Lock()
defer h.mu.Unlock()
for _, conns := range h.clients {
for c := range conns {
_ = c.Conn.Close()
}
}
}
// helper
func toInt64(v interface{}) int64 {
switch x := v.(type) {
case float64:
return int64(x)
case int64:
return x
case int:
return int64(x)
case string:
i, _ := strconv.ParseInt(x, 10, 64)
return i
}
return 0
}
func toStringSlice(v interface{}) []string {
arr, ok := v.([]interface{})
if !ok {
return nil
}
out := make([]string, 0, len(arr))
for _, item := range arr {
if s, ok := item.(string); ok {
out = append(out, s)
}
}
return out
}

52
backend/migrations/2026_06_22_012_activity_messages.sql Normal file
View File

@ -0,0 +1,52 @@
-- 2026_06_22_012_activity_messages.sql
BEGIN;
CREATE TABLE IF NOT EXISTS public.activity_messages (
id BIGSERIAL PRIMARY KEY,
activity_id BIGINT NOT NULL,
user_id BIGINT NOT NULL,
star_id BIGINT NOT NULL,
nickname VARCHAR(50), -- 缓存昵称（写入时回查，避免读时 RPC 反查）
avatar_url VARCHAR(500), -- 缓存头像 URL
content VARCHAR(500) NOT NULL,
status SMALLINT NOT NULL DEFAULT 0,
created_at BIGINT NOT NULL,
updated_at BIGINT NOT NULL,
deleted_at TIMESTAMP WITH TIME ZONE, -- gorm.DeletedAt 软删除时间戳
CONSTRAINT fk_messages_activity
FOREIGN KEY (activity_id) REFERENCES public.activities(id) ON DELETE CASCADE,
CONSTRAINT fk_messages_user
FOREIGN KEY (user_id) REFERENCES public.users(id),
CONSTRAINT fk_messages_star
FOREIGN KEY (star_id) REFERENCES public.stars(star_id),
CONSTRAINT chk_activity_messages_status CHECK (status BETWEEN 0 AND 2)
);
ALTER SEQUENCE activity_messages_id_seq RESTART WITH 10000;
CREATE INDEX IF NOT EXISTS idx_activity_messages_activity_created
ON public.activity_messages (activity_id, created_at DESC, id DESC)
WHERE deleted_at IS NULL;
CREATE INDEX IF NOT EXISTS idx_activity_messages_user_created
ON public.activity_messages (user_id, created_at DESC)
WHERE deleted_at IS NULL;
CREATE INDEX IF NOT EXISTS idx_activity_messages_activity_incr
ON public.activity_messages (activity_id, id DESC)
WHERE deleted_at IS NULL;
COMMENT ON TABLE public.activity_messages IS '活动留言表';
COMMENT ON COLUMN public.activity_messages.id IS '主键，自增';
COMMENT ON COLUMN public.activity_messages.activity_id IS 'FK -> activities.id';
COMMENT ON COLUMN public.activity_messages.user_id IS '留言用户 ID';
COMMENT ON COLUMN public.activity_messages.star_id IS '所属明星/星球 ID';
COMMENT ON COLUMN public.activity_messages.nickname IS '缓存昵称（写入时回查，避免读时 RPC 反查）';
COMMENT ON COLUMN public.activity_messages.avatar_url IS '缓存头像 URL';
COMMENT ON COLUMN public.activity_messages.content IS '留言正文，1-500 字';
COMMENT ON COLUMN public.activity_messages.status IS '0=正常|1=隐藏|2=已删除';
COMMENT ON COLUMN public.activity_messages.created_at IS '留言时间，毫秒时间戳';
COMMENT ON COLUMN public.activity_messages.updated_at IS '更新时间，毫秒时间戳';
COMMENT ON COLUMN public.activity_messages.deleted_at IS '软删除时间戳（gorm.DeletedAt，NULL=未删除）';
COMMIT;

26
backend/pkg/errors/errors.go
View File

@ -68,6 +68,15 @@ var (
ErrActivityNotFound = errors.New("活动不存在")
ErrActivityItemNotFound = errors.New("活动道具不存在")
// 活动留言相关错误
ErrActivityMessageNotFound = errors.New("活动留言不存在")
ErrActivityMessageTooFrequent = errors.New("留言太频繁，请稍后再试")
ErrActivityMessageLimitReached = errors.New("当前活动留言已达上限")
ErrActivityMessageContentEmpty = errors.New("留言内容不能为空")
ErrActivityMessageContentTooLong = errors.New("留言内容过长，最多500字")
ErrActivityMessageContentInvalid = errors.New("留言内容包含不当内容")
ErrActivityMessageActivityInactive = errors.New("活动不在进行中")
// 星册服务相关错误
ErrCollectionAssetNotFound = errors.New("典藏藏品不存在")
ErrActivityAssetNotFound = errors.New("活动藏品不存在")
@ -122,6 +131,23 @@ func ToGRPCCode(err error) codes.Code {
return codes.NotFound
case errors.Is(err, ErrInvalidAssetType):
return codes.InvalidArgument
case errors.Is(err, ErrActivityMessageNotFound):
return codes.NotFound
case errors.Is(err, ErrActivityMessageTooFrequent):
return codes.ResourceExhausted
case errors.Is(err, ErrActivityMessageLimitReached):
return codes.ResourceExhausted
case errors.Is(err, ErrActivityMessageContentEmpty):
return codes.InvalidArgument
case errors.Is(err, ErrActivityMessageContentTooLong):
return codes.InvalidArgument
case errors.Is(err, ErrActivityMessageContentInvalid):
// 内容含敏感词 → 参数级校验错(3);不是权限不足,不能映射到 PermissionDenied(7),
// 否则前端 api.js 会把它当成"账号被封"清缓存跳登录页。
return codes.InvalidArgument
case errors.Is(err, ErrActivityMessageActivityInactive):
// 活动未在进行中 → 系统前置条件不满足(9);同样避免 PermissionDenied(7) 的副作用。
return codes.FailedPrecondition
default:
return codes.Internal
}

23
backend/pkg/models/activity_message.go Normal file
View File

@ -0,0 +1,23 @@
package models
import "gorm.io/gorm"
// ActivityMessage 活动留言
type ActivityMessage struct {
ID int64 `json:"id" gorm:"primaryKey;autoIncrement"`
ActivityID int64 `json:"activity_id" gorm:"not null;index"`
UserID int64 `json:"user_id" gorm:"not null;index"`
StarID int64 `json:"star_id" gorm:"not null"`
Nickname string `json:"nickname" gorm:"size:50"`
AvatarURL string `json:"avatar_url" gorm:"size:500"`
Content string `json:"content" gorm:"type:varchar(500);not null"`
Status int16 `json:"status" gorm:"default:0"`
CreatedAt int64 `json:"created_at" gorm:"not null"`
UpdatedAt int64 `json:"updated_at" gorm:"not null"`
DeletedAt gorm.DeletedAt `json:"deleted_at,omitzero" gorm:"column:deleted_at;index"`
}
// TableName 表名
func (ActivityMessage) TableName() string {
return "activity_messages"
}

472
backend/pkg/proto/activity/activity.pb.go
View File

@ -1898,6 +1898,362 @@ func (x *GetLatestContributionsResponse) GetRecords() []*ContributionRecord {
return nil
}
type ActivityMessage struct {
state protoimpl.MessageState `protogen:"open.v1"`
Id int64 `protobuf:"varint,1,opt,name=id,proto3" json:"id,omitempty"`
ActivityId int64 `protobuf:"varint,2,opt,name=activity_id,json=activityId,proto3" json:"activity_id,omitempty"`
UserId int64 `protobuf:"varint,3,opt,name=user_id,json=userId,proto3" json:"user_id,omitempty"`
StarId int64 `protobuf:"varint,4,opt,name=star_id,json=starId,proto3" json:"star_id,omitempty"`
Nickname string `protobuf:"bytes,5,opt,name=nickname,proto3" json:"nickname,omitempty"`
AvatarUrl string `protobuf:"bytes,6,opt,name=avatar_url,json=avatarUrl,proto3" json:"avatar_url,omitempty"`
Content string `protobuf:"bytes,7,opt,name=content,proto3" json:"content,omitempty"`
CreatedAt int64 `protobuf:"varint,8,opt,name=created_at,json=createdAt,proto3" json:"created_at,omitempty"`
unknownFields protoimpl.UnknownFields
sizeCache protoimpl.SizeCache
}
func (x *ActivityMessage) Reset() {
*x = ActivityMessage{}
mi := &file_activity_proto_msgTypes[23]
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
ms.StoreMessageInfo(mi)
}
func (x *ActivityMessage) String() string {
return protoimpl.X.MessageStringOf(x)
}
func (*ActivityMessage) ProtoMessage() {}
func (x *ActivityMessage) ProtoReflect() protoreflect.Message {
mi := &file_activity_proto_msgTypes[23]
if x != nil {
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
if ms.LoadMessageInfo() == nil {
ms.StoreMessageInfo(mi)
}
return ms
}
return mi.MessageOf(x)
}
// Deprecated: Use ActivityMessage.ProtoReflect.Descriptor instead.
func (*ActivityMessage) Descriptor() ([]byte, []int) {
return file_activity_proto_rawDescGZIP(), []int{23}
}
func (x *ActivityMessage) GetId() int64 {
if x != nil {
return x.Id
}
return 0
}
func (x *ActivityMessage) GetActivityId() int64 {
if x != nil {
return x.ActivityId
}
return 0
}
func (x *ActivityMessage) GetUserId() int64 {
if x != nil {
return x.UserId
}
return 0
}
func (x *ActivityMessage) GetStarId() int64 {
if x != nil {
return x.StarId
}
return 0
}
func (x *ActivityMessage) GetNickname() string {
if x != nil {
return x.Nickname
}
return ""
}
func (x *ActivityMessage) GetAvatarUrl() string {
if x != nil {
return x.AvatarUrl
}
return ""
}
func (x *ActivityMessage) GetContent() string {
if x != nil {
return x.Content
}
return ""
}
func (x *ActivityMessage) GetCreatedAt() int64 {
if x != nil {
return x.CreatedAt
}
return 0
}
type ListActivityMessagesRequest struct {
state protoimpl.MessageState `protogen:"open.v1"`
ActivityId int64 `protobuf:"varint,1,opt,name=activity_id,json=activityId,proto3" json:"activity_id,omitempty"`
Page int32 `protobuf:"varint,2,opt,name=page,proto3" json:"page,omitempty"`
PageSize int32 `protobuf:"varint,3,opt,name=page_size,json=pageSize,proto3" json:"page_size,omitempty"`
unknownFields protoimpl.UnknownFields
sizeCache protoimpl.SizeCache
}
func (x *ListActivityMessagesRequest) Reset() {
*x = ListActivityMessagesRequest{}
mi := &file_activity_proto_msgTypes[24]
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
ms.StoreMessageInfo(mi)
}
func (x *ListActivityMessagesRequest) String() string {
return protoimpl.X.MessageStringOf(x)
}
func (*ListActivityMessagesRequest) ProtoMessage() {}
func (x *ListActivityMessagesRequest) ProtoReflect() protoreflect.Message {
mi := &file_activity_proto_msgTypes[24]
if x != nil {
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
if ms.LoadMessageInfo() == nil {
ms.StoreMessageInfo(mi)
}
return ms
}
return mi.MessageOf(x)
}
// Deprecated: Use ListActivityMessagesRequest.ProtoReflect.Descriptor instead.
func (*ListActivityMessagesRequest) Descriptor() ([]byte, []int) {
return file_activity_proto_rawDescGZIP(), []int{24}
}
func (x *ListActivityMessagesRequest) GetActivityId() int64 {
if x != nil {
return x.ActivityId
}
return 0
}
func (x *ListActivityMessagesRequest) GetPage() int32 {
if x != nil {
return x.Page
}
return 0
}
func (x *ListActivityMessagesRequest) GetPageSize() int32 {
if x != nil {
return x.PageSize
}
return 0
}
type ListActivityMessagesResponse struct {
state protoimpl.MessageState `protogen:"open.v1"`
Base *common.BaseResponse `protobuf:"bytes,1,opt,name=base,proto3" json:"base,omitempty"`
Messages []*ActivityMessage `protobuf:"bytes,2,rep,name=messages,proto3" json:"messages,omitempty"`
Page int32 `protobuf:"varint,3,opt,name=page,proto3" json:"page,omitempty"`
PageSize int32 `protobuf:"varint,4,opt,name=page_size,json=pageSize,proto3" json:"page_size,omitempty"`
Total int32 `protobuf:"varint,5,opt,name=total,proto3" json:"total,omitempty"`
unknownFields protoimpl.UnknownFields
sizeCache protoimpl.SizeCache
}
func (x *ListActivityMessagesResponse) Reset() {
*x = ListActivityMessagesResponse{}
mi := &file_activity_proto_msgTypes[25]
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
ms.StoreMessageInfo(mi)
}
func (x *ListActivityMessagesResponse) String() string {
return protoimpl.X.MessageStringOf(x)
}
func (*ListActivityMessagesResponse) ProtoMessage() {}
func (x *ListActivityMessagesResponse) ProtoReflect() protoreflect.Message {
mi := &file_activity_proto_msgTypes[25]
if x != nil {
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
if ms.LoadMessageInfo() == nil {
ms.StoreMessageInfo(mi)
}
return ms
}
return mi.MessageOf(x)
}
// Deprecated: Use ListActivityMessagesResponse.ProtoReflect.Descriptor instead.
func (*ListActivityMessagesResponse) Descriptor() ([]byte, []int) {
return file_activity_proto_rawDescGZIP(), []int{25}
}
func (x *ListActivityMessagesResponse) GetBase() *common.BaseResponse {
if x != nil {
return x.Base
}
return nil
}
func (x *ListActivityMessagesResponse) GetMessages() []*ActivityMessage {
if x != nil {
return x.Messages
}
return nil
}
func (x *ListActivityMessagesResponse) GetPage() int32 {
if x != nil {
return x.Page
}
return 0
}
func (x *ListActivityMessagesResponse) GetPageSize() int32 {
if x != nil {
return x.PageSize
}
return 0
}
func (x *ListActivityMessagesResponse) GetTotal() int32 {
if x != nil {
return x.Total
}
return 0
}
type CreateActivityMessageRequest struct {
state protoimpl.MessageState `protogen:"open.v1"`
ActivityId int64 `protobuf:"varint,1,opt,name=activity_id,json=activityId,proto3" json:"activity_id,omitempty"`
UserId int64 `protobuf:"varint,2,opt,name=user_id,json=userId,proto3" json:"user_id,omitempty"`
StarId int64 `protobuf:"varint,3,opt,name=star_id,json=starId,proto3" json:"star_id,omitempty"`
Content string `protobuf:"bytes,4,opt,name=content,proto3" json:"content,omitempty"`
unknownFields protoimpl.UnknownFields
sizeCache protoimpl.SizeCache
}
func (x *CreateActivityMessageRequest) Reset() {
*x = CreateActivityMessageRequest{}
mi := &file_activity_proto_msgTypes[26]
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
ms.StoreMessageInfo(mi)
}
func (x *CreateActivityMessageRequest) String() string {
return protoimpl.X.MessageStringOf(x)
}
func (*CreateActivityMessageRequest) ProtoMessage() {}
func (x *CreateActivityMessageRequest) ProtoReflect() protoreflect.Message {
mi := &file_activity_proto_msgTypes[26]
if x != nil {
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
if ms.LoadMessageInfo() == nil {
ms.StoreMessageInfo(mi)
}
return ms
}
return mi.MessageOf(x)
}
// Deprecated: Use CreateActivityMessageRequest.ProtoReflect.Descriptor instead.
func (*CreateActivityMessageRequest) Descriptor() ([]byte, []int) {
return file_activity_proto_rawDescGZIP(), []int{26}
}
func (x *CreateActivityMessageRequest) GetActivityId() int64 {
if x != nil {
return x.ActivityId
}
return 0
}
func (x *CreateActivityMessageRequest) GetUserId() int64 {
if x != nil {
return x.UserId
}
return 0
}
func (x *CreateActivityMessageRequest) GetStarId() int64 {
if x != nil {
return x.StarId
}
return 0
}
func (x *CreateActivityMessageRequest) GetContent() string {
if x != nil {
return x.Content
}
return ""
}
type CreateActivityMessageResponse struct {
state protoimpl.MessageState `protogen:"open.v1"`
Base *common.BaseResponse `protobuf:"bytes,1,opt,name=base,proto3" json:"base,omitempty"`
Message *ActivityMessage `protobuf:"bytes,2,opt,name=message,proto3" json:"message,omitempty"`
unknownFields protoimpl.UnknownFields
sizeCache protoimpl.SizeCache
}
func (x *CreateActivityMessageResponse) Reset() {
*x = CreateActivityMessageResponse{}
mi := &file_activity_proto_msgTypes[27]
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
ms.StoreMessageInfo(mi)
}
func (x *CreateActivityMessageResponse) String() string {
return protoimpl.X.MessageStringOf(x)
}
func (*CreateActivityMessageResponse) ProtoMessage() {}
func (x *CreateActivityMessageResponse) ProtoReflect() protoreflect.Message {
mi := &file_activity_proto_msgTypes[27]
if x != nil {
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
if ms.LoadMessageInfo() == nil {
ms.StoreMessageInfo(mi)
}
return ms
}
return mi.MessageOf(x)
}
// Deprecated: Use CreateActivityMessageResponse.ProtoReflect.Descriptor instead.
func (*CreateActivityMessageResponse) Descriptor() ([]byte, []int) {
return file_activity_proto_rawDescGZIP(), []int{27}
}
func (x *CreateActivityMessageResponse) GetBase() *common.BaseResponse {
if x != nil {
return x.Base
}
return nil
}
func (x *CreateActivityMessageResponse) GetMessage() *ActivityMessage {
if x != nil {
return x.Message
}
return nil
}
var File_activity_proto protoreflect.FileDescriptor
const file_activity_proto_rawDesc = "" +
@ -2077,8 +2433,39 @@ const file_activity_proto_rawDesc = "" +
"created_at\x18\f \x01(\x03R\tcreatedAt\"\x92\x01\n" +
"\x1eGetLatestContributionsResponse\x120\n" +
"\x04base\x18\x01 \x01(\v2\x1c.topfans.common.BaseResponseR\x04base\x12>\n" +
"\arecords\x18\x02 \x03(\v2$.topfans.activity.ContributionRecordR\arecords2\xf9\n" +
"\arecords\x18\x02 \x03(\v2$.topfans.activity.ContributionRecordR\arecords\"\xe8\x01\n" +
"\x0fActivityMessage\x12\x0e\n" +
"\x02id\x18\x01 \x01(\x03R\x02id\x12\x1f\n" +
"\vactivity_id\x18\x02 \x01(\x03R\n" +
"activityId\x12\x17\n" +
"\auser_id\x18\x03 \x01(\x03R\x06userId\x12\x17\n" +
"\astar_id\x18\x04 \x01(\x03R\x06starId\x12\x1a\n" +
"\bnickname\x18\x05 \x01(\tR\bnickname\x12\x1d\n" +
"\n" +
"avatar_url\x18\x06 \x01(\tR\tavatarUrl\x12\x18\n" +
"\acontent\x18\a \x01(\tR\acontent\x12\x1d\n" +
"\n" +
"created_at\x18\b \x01(\x03R\tcreatedAt\"o\n" +
"\x1bListActivityMessagesRequest\x12\x1f\n" +
"\vactivity_id\x18\x01 \x01(\x03R\n" +
"activityId\x12\x12\n" +
"\x04page\x18\x02 \x01(\x05R\x04page\x12\x1b\n" +
"\tpage_size\x18\x03 \x01(\x05R\bpageSize\"\xd6\x01\n" +
"\x1cListActivityMessagesResponse\x120\n" +
"\x04base\x18\x01 \x01(\v2\x1c.topfans.common.BaseResponseR\x04base\x12=\n" +
"\bmessages\x18\x02 \x03(\v2!.topfans.activity.ActivityMessageR\bmessages\x12\x12\n" +
"\x04page\x18\x03 \x01(\x05R\x04page\x12\x1b\n" +
"\tpage_size\x18\x04 \x01(\x05R\bpageSize\x12\x14\n" +
"\x05total\x18\x05 \x01(\x05R\x05total\"\x8b\x01\n" +
"\x1cCreateActivityMessageRequest\x12\x1f\n" +
"\vactivity_id\x18\x01 \x01(\x03R\n" +
"activityId\x12\x17\n" +
"\auser_id\x18\x02 \x01(\x03R\x06userId\x12\x17\n" +
"\astar_id\x18\x03 \x01(\x03R\x06starId\x12\x18\n" +
"\acontent\x18\x04 \x01(\tR\acontent\"\x8e\x01\n" +
"\x1dCreateActivityMessageResponse\x120\n" +
"\x04base\x18\x01 \x01(\v2\x1c.topfans.common.BaseResponseR\x04base\x12;\n" +
"\amessage\x18\x02 \x01(\v2!.topfans.activity.ActivityMessageR\amessage2\xd5\r\n" +
"\x0fActivityService\x12\x82\x01\n" +
"\x0fGetActivityList\x12(.topfans.activity.GetActivityListRequest\x1a).topfans.activity.GetActivityListResponse\"\x1a\x82\xd3\xe4\x93\x02\x14\x12\x12/api/v1/activities\x12y\n" +
"\vGetActivity\x12$.topfans.activity.GetProgressRequest\x1a\x1a.topfans.activity.Activity\"(\x82\xd3\xe4\x93\x02\"\x12 /api/v1/activities/{activity_id}\x12\x91\x01\n" +
@ -2088,7 +2475,9 @@ const file_activity_proto_rawDesc = "" +
"\x11BatchPurchaseItem\x12*.topfans.activity.BatchPurchaseItemRequest\x1a+.topfans.activity.BatchPurchaseItemResponse\":\x82\xd3\xe4\x93\x024:\x01*\"//api/v1/activities/{activity_id}/batch-purchase\x12\xa7\x01\n" +
"\x16GetContributionRanking\x12,.topfans.activity.ContributionRankingRequest\x1a-.topfans.activity.ContributionRankingResponse\"0\x82\xd3\xe4\x93\x02*\x12(/api/v1/activities/{activity_id}/ranking\x12\x99\x01\n" +
"\x14GetMintingActivities\x12-.topfans.activity.GetMintingActivitiesRequest\x1a..topfans.activity.GetMintingActivitiesResponse\"\"\x82\xd3\xe4\x93\x02\x1c\x12\x1a/api/v1/minting-activities\x12\xba\x01\n" +
"\x16GetLatestContributions\x12/.topfans.activity.GetLatestContributionsRequest\x1a0.topfans.activity.GetLatestContributionsResponse\"=\x82\xd3\xe4\x93\x027\x125/api/v1/activities/{activity_id}/contributions/latestB8Z6github.com/topfans/backend/pkg/proto/activity;activityb\x06proto3"
"\x16GetLatestContributions\x12/.topfans.activity.GetLatestContributionsRequest\x1a0.topfans.activity.GetLatestContributionsResponse\"=\x82\xd3\xe4\x93\x027\x125/api/v1/activities/{activity_id}/contributions/latest\x12\xa8\x01\n" +
"\x14ListActivityMessages\x12-.topfans.activity.ListActivityMessagesRequest\x1a..topfans.activity.ListActivityMessagesResponse\"1\x82\xd3\xe4\x93\x02+\x12)/api/v1/activities/{activity_id}/messages\x12\xae\x01\n" +
"\x15CreateActivityMessage\x12..topfans.activity.CreateActivityMessageRequest\x1a/.topfans.activity.CreateActivityMessageResponse\"4\x82\xd3\xe4\x93\x02.:\x01*\")/api/v1/activities/{activity_id}/messagesB8Z6github.com/topfans/backend/pkg/proto/activity;activityb\x06proto3"
var (
file_activity_proto_rawDescOnce sync.Once
@ -2102,7 +2491,7 @@ func file_activity_proto_rawDescGZIP() []byte {
return file_activity_proto_rawDescData
}
var file_activity_proto_msgTypes = make([]protoimpl.MessageInfo, 23)
var file_activity_proto_msgTypes = make([]protoimpl.MessageInfo, 28)
var file_activity_proto_goTypes = []any{
(*Activity)(nil), // 0: topfans.activity.Activity
(*ActivityItem)(nil), // 1: topfans.activity.ActivityItem
@ -2127,48 +2516,61 @@ var file_activity_proto_goTypes = []any{
(*GetLatestContributionsRequest)(nil), // 20: topfans.activity.GetLatestContributionsRequest
(*ContributionRecord)(nil), // 21: topfans.activity.ContributionRecord
(*GetLatestContributionsResponse)(nil), // 22: topfans.activity.GetLatestContributionsResponse
(*common.BaseResponse)(nil), // 23: topfans.common.BaseResponse
(*ActivityMessage)(nil), // 23: topfans.activity.ActivityMessage
(*ListActivityMessagesRequest)(nil), // 24: topfans.activity.ListActivityMessagesRequest
(*ListActivityMessagesResponse)(nil), // 25: topfans.activity.ListActivityMessagesResponse
(*CreateActivityMessageRequest)(nil), // 26: topfans.activity.CreateActivityMessageRequest
(*CreateActivityMessageResponse)(nil), // 27: topfans.activity.CreateActivityMessageResponse
(*common.BaseResponse)(nil), // 28: topfans.common.BaseResponse
}
var file_activity_proto_depIdxs = []int32{
1, // 0: topfans.activity.Activity.items:type_name -> topfans.activity.ActivityItem
1, // 1: topfans.activity.ActivityItemsResponse.items:type_name -> topfans.activity.ActivityItem
23, // 2: topfans.activity.PurchaseItemResponse.base:type_name -> topfans.common.BaseResponse
28, // 2: topfans.activity.PurchaseItemResponse.base:type_name -> topfans.common.BaseResponse
5, // 3: topfans.activity.BatchPurchaseItemRequest.items:type_name -> topfans.activity.PurchaseItem
23, // 4: topfans.activity.BatchPurchaseItemResponse.base:type_name -> topfans.common.BaseResponse
28, // 4: topfans.activity.BatchPurchaseItemResponse.base:type_name -> topfans.common.BaseResponse
8, // 5: topfans.activity.BatchPurchaseItemResponse.fails:type_name -> topfans.activity.PurchaseFailItem
23, // 6: topfans.activity.ContributionRankingResponse.base:type_name -> topfans.common.BaseResponse
28, // 6: topfans.activity.ContributionRankingResponse.base:type_name -> topfans.common.BaseResponse
10, // 7: topfans.activity.ContributionRankingResponse.items:type_name -> topfans.activity.ContributionRankingItem
12, // 8: topfans.activity.ContributionRankingResponse.my_contribution:type_name -> topfans.activity.MyContribution
23, // 9: topfans.activity.GetActivityListResponse.base:type_name -> topfans.common.BaseResponse
28, // 9: topfans.activity.GetActivityListResponse.base:type_name -> topfans.common.BaseResponse
0, // 10: topfans.activity.GetActivityListResponse.activities:type_name -> topfans.activity.Activity
23, // 11: topfans.activity.GetProgressResponse.base:type_name -> topfans.common.BaseResponse
23, // 12: topfans.activity.GetMintingActivitiesResponse.base:type_name -> topfans.common.BaseResponse
28, // 11: topfans.activity.GetProgressResponse.base:type_name -> topfans.common.BaseResponse
28, // 12: topfans.activity.GetMintingActivitiesResponse.base:type_name -> topfans.common.BaseResponse
17, // 13: topfans.activity.GetMintingActivitiesResponse.activities:type_name -> topfans.activity.MintingActivity
23, // 14: topfans.activity.GetLatestContributionsResponse.base:type_name -> topfans.common.BaseResponse
28, // 14: topfans.activity.GetLatestContributionsResponse.base:type_name -> topfans.common.BaseResponse
21, // 15: topfans.activity.GetLatestContributionsResponse.records:type_name -> topfans.activity.ContributionRecord
13, // 16: topfans.activity.ActivityService.GetActivityList:input_type -> topfans.activity.GetActivityListRequest
15, // 17: topfans.activity.ActivityService.GetActivity:input_type -> topfans.activity.GetProgressRequest
15, // 18: topfans.activity.ActivityService.GetActivityItems:input_type -> topfans.activity.GetProgressRequest
15, // 19: topfans.activity.ActivityService.GetProgress:input_type -> topfans.activity.GetProgressRequest
3, // 20: topfans.activity.ActivityService.PurchaseItem:input_type -> topfans.activity.PurchaseItemRequest
6, // 21: topfans.activity.ActivityService.BatchPurchaseItem:input_type -> topfans.activity.BatchPurchaseItemRequest
9, // 22: topfans.activity.ActivityService.GetContributionRanking:input_type -> topfans.activity.ContributionRankingRequest
18, // 23: topfans.activity.ActivityService.GetMintingActivities:input_type -> topfans.activity.GetMintingActivitiesRequest
20, // 24: topfans.activity.ActivityService.GetLatestContributions:input_type -> topfans.activity.GetLatestContributionsRequest
14, // 25: topfans.activity.ActivityService.GetActivityList:output_type -> topfans.activity.GetActivityListResponse
0, // 26: topfans.activity.ActivityService.GetActivity:output_type -> topfans.activity.Activity
2, // 27: topfans.activity.ActivityService.GetActivityItems:output_type -> topfans.activity.ActivityItemsResponse
16, // 28: topfans.activity.ActivityService.GetProgress:output_type -> topfans.activity.GetProgressResponse
4, // 29: topfans.activity.ActivityService.PurchaseItem:output_type -> topfans.activity.PurchaseItemResponse
7, // 30: topfans.activity.ActivityService.BatchPurchaseItem:output_type -> topfans.activity.BatchPurchaseItemResponse
11, // 31: topfans.activity.ActivityService.GetContributionRanking:output_type -> topfans.activity.ContributionRankingResponse
19, // 32: topfans.activity.ActivityService.GetMintingActivities:output_type -> topfans.activity.GetMintingActivitiesResponse
22, // 33: topfans.activity.ActivityService.GetLatestContributions:output_type -> topfans.activity.GetLatestContributionsResponse
25, // [25:34] is the sub-list for method output_type
16, // [16:25] is the sub-list for method input_type
16, // [16:16] is the sub-list for extension type_name
16, // [16:16] is the sub-list for extension extendee
0, // [0:16] is the sub-list for field type_name
28, // 16: topfans.activity.ListActivityMessagesResponse.base:type_name -> topfans.common.BaseResponse
23, // 17: topfans.activity.ListActivityMessagesResponse.messages:type_name -> topfans.activity.ActivityMessage
28, // 18: topfans.activity.CreateActivityMessageResponse.base:type_name -> topfans.common.BaseResponse
23, // 19: topfans.activity.CreateActivityMessageResponse.message:type_name -> topfans.activity.ActivityMessage
13, // 20: topfans.activity.ActivityService.GetActivityList:input_type -> topfans.activity.GetActivityListRequest
15, // 21: topfans.activity.ActivityService.GetActivity:input_type -> topfans.activity.GetProgressRequest
15, // 22: topfans.activity.ActivityService.GetActivityItems:input_type -> topfans.activity.GetProgressRequest
15, // 23: topfans.activity.ActivityService.GetProgress:input_type -> topfans.activity.GetProgressRequest
3, // 24: topfans.activity.ActivityService.PurchaseItem:input_type -> topfans.activity.PurchaseItemRequest
6, // 25: topfans.activity.ActivityService.BatchPurchaseItem:input_type -> topfans.activity.BatchPurchaseItemRequest
9, // 26: topfans.activity.ActivityService.GetContributionRanking:input_type -> topfans.activity.ContributionRankingRequest
18, // 27: topfans.activity.ActivityService.GetMintingActivities:input_type -> topfans.activity.GetMintingActivitiesRequest
20, // 28: topfans.activity.ActivityService.GetLatestContributions:input_type -> topfans.activity.GetLatestContributionsRequest
24, // 29: topfans.activity.ActivityService.ListActivityMessages:input_type -> topfans.activity.ListActivityMessagesRequest
26, // 30: topfans.activity.ActivityService.CreateActivityMessage:input_type -> topfans.activity.CreateActivityMessageRequest
14, // 31: topfans.activity.ActivityService.GetActivityList:output_type -> topfans.activity.GetActivityListResponse
0, // 32: topfans.activity.ActivityService.GetActivity:output_type -> topfans.activity.Activity
2, // 33: topfans.activity.ActivityService.GetActivityItems:output_type -> topfans.activity.ActivityItemsResponse
16, // 34: topfans.activity.ActivityService.GetProgress:output_type -> topfans.activity.GetProgressResponse
4, // 35: topfans.activity.ActivityService.PurchaseItem:output_type -> topfans.activity.PurchaseItemResponse
7, // 36: topfans.activity.ActivityService.BatchPurchaseItem:output_type -> topfans.activity.BatchPurchaseItemResponse
11, // 37: topfans.activity.ActivityService.GetContributionRanking:output_type -> topfans.activity.ContributionRankingResponse
19, // 38: topfans.activity.ActivityService.GetMintingActivities:output_type -> topfans.activity.GetMintingActivitiesResponse
22, // 39: topfans.activity.ActivityService.GetLatestContributions:output_type -> topfans.activity.GetLatestContributionsResponse
25, // 40: topfans.activity.ActivityService.ListActivityMessages:output_type -> topfans.activity.ListActivityMessagesResponse
27, // 41: topfans.activity.ActivityService.CreateActivityMessage:output_type -> topfans.activity.CreateActivityMessageResponse
31, // [31:42] is the sub-list for method output_type
20, // [20:31] is the sub-list for method input_type
20, // [20:20] is the sub-list for extension type_name
20, // [20:20] is the sub-list for extension extendee
0, // [0:20] is the sub-list for field type_name
}
func init() { file_activity_proto_init() }
@ -2182,7 +2584,7 @@ func file_activity_proto_init() {
GoPackagePath: reflect.TypeOf(x{}).PkgPath(),
RawDescriptor: unsafe.Slice(unsafe.StringData(file_activity_proto_rawDesc), len(file_activity_proto_rawDesc)),
NumEnums: 0,
NumMessages: 23,
NumMessages: 28,
NumExtensions: 0,
NumServices: 1,
},

56
backend/pkg/proto/activity/activity.triple.go
View File

@ -54,6 +54,10 @@ const (
ActivityServiceGetMintingActivitiesProcedure = "/topfans.activity.ActivityService/GetMintingActivities"
// ActivityServiceGetLatestContributionsProcedure is the fully-qualified name of the ActivityService's GetLatestContributions RPC.
ActivityServiceGetLatestContributionsProcedure = "/topfans.activity.ActivityService/GetLatestContributions"
// ActivityServiceListActivityMessagesProcedure is the fully-qualified name of the ActivityService's ListActivityMessages RPC.
ActivityServiceListActivityMessagesProcedure = "/topfans.activity.ActivityService/ListActivityMessages"
// ActivityServiceCreateActivityMessageProcedure is the fully-qualified name of the ActivityService's CreateActivityMessage RPC.
ActivityServiceCreateActivityMessageProcedure = "/topfans.activity.ActivityService/CreateActivityMessage"
)
var (
@ -71,6 +75,8 @@ type ActivityService interface {
GetContributionRanking(ctx context.Context, req *ContributionRankingRequest, opts ...client.CallOption) (*ContributionRankingResponse, error)
GetMintingActivities(ctx context.Context, req *GetMintingActivitiesRequest, opts ...client.CallOption) (*GetMintingActivitiesResponse, error)
GetLatestContributions(ctx context.Context, req *GetLatestContributionsRequest, opts ...client.CallOption) (*GetLatestContributionsResponse, error)
ListActivityMessages(ctx context.Context, req *ListActivityMessagesRequest, opts ...client.CallOption) (*ListActivityMessagesResponse, error)
CreateActivityMessage(ctx context.Context, req *CreateActivityMessageRequest, opts ...client.CallOption) (*CreateActivityMessageResponse, error)
}
// NewActivityService constructs a client for the activity.ActivityService service.
@ -165,9 +171,25 @@ func (c *ActivityServiceImpl) GetLatestContributions(ctx context.Context, req *G
return resp, nil
}
func (c *ActivityServiceImpl) ListActivityMessages(ctx context.Context, req *ListActivityMessagesRequest, opts ...client.CallOption) (*ListActivityMessagesResponse, error) {
resp := new(ListActivityMessagesResponse)
if err := c.conn.CallUnary(ctx, []interface{}{req}, resp, "ListActivityMessages", opts...); err != nil {
return nil, err
}
return resp, nil
}
func (c *ActivityServiceImpl) CreateActivityMessage(ctx context.Context, req *CreateActivityMessageRequest, opts ...client.CallOption) (*CreateActivityMessageResponse, error) {
resp := new(CreateActivityMessageResponse)
if err := c.conn.CallUnary(ctx, []interface{}{req}, resp, "CreateActivityMessage", opts...); err != nil {
return nil, err
}
return resp, nil
}
var ActivityService_ClientInfo = client.ClientInfo{
InterfaceName: "topfans.activity.ActivityService",
MethodNames: []string{"GetActivityList", "GetActivity", "GetActivityItems", "GetProgress", "PurchaseItem", "BatchPurchaseItem", "GetContributionRanking", "GetMintingActivities", "GetLatestContributions"},
MethodNames: []string{"GetActivityList", "GetActivity", "GetActivityItems", "GetProgress", "PurchaseItem", "BatchPurchaseItem", "GetContributionRanking", "GetMintingActivities", "GetLatestContributions", "ListActivityMessages", "CreateActivityMessage"},
ConnectionInjectFunc: func(dubboCliRaw interface{}, conn *client.Connection) {
dubboCli := dubboCliRaw.(*ActivityServiceImpl)
dubboCli.conn = conn
@ -185,6 +207,8 @@ type ActivityServiceHandler interface {
GetContributionRanking(context.Context, *ContributionRankingRequest) (*ContributionRankingResponse, error)
GetMintingActivities(context.Context, *GetMintingActivitiesRequest) (*GetMintingActivitiesResponse, error)
GetLatestContributions(context.Context, *GetLatestContributionsRequest) (*GetLatestContributionsResponse, error)
ListActivityMessages(context.Context, *ListActivityMessagesRequest) (*ListActivityMessagesResponse, error)
CreateActivityMessage(context.Context, *CreateActivityMessageRequest) (*CreateActivityMessageResponse, error)
}
func RegisterActivityServiceHandler(srv *server.Server, hdlr ActivityServiceHandler, opts ...server.ServiceOption) error {
@ -334,5 +358,35 @@ var ActivityService_ServiceInfo = server.ServiceInfo{
return triple_protocol.NewResponse(res), nil
},
},
{
Name: "ListActivityMessages",
Type: constant.CallUnary,
ReqInitFunc: func() interface{} {
return new(ListActivityMessagesRequest)
},
MethodFunc: func(ctx context.Context, args []interface{}, handler interface{}) (interface{}, error) {
req := args[0].(*ListActivityMessagesRequest)
res, err := handler.(ActivityServiceHandler).ListActivityMessages(ctx, req)
if err != nil {
return nil, err
}
return triple_protocol.NewResponse(res), nil
},
},
{
Name: "CreateActivityMessage",
Type: constant.CallUnary,
ReqInitFunc: func() interface{} {
return new(CreateActivityMessageRequest)
},
MethodFunc: func(ctx context.Context, args []interface{}, handler interface{}) (interface{}, error) {
req := args[0].(*CreateActivityMessageRequest)
res, err := handler.(ActivityServiceHandler).CreateActivityMessage(ctx, req)
if err != nil {
return nil, err
}
return triple_protocol.NewResponse(res), nil
},
},
},
}

52
backend/proto/activity.proto
View File

@ -231,6 +231,43 @@ message GetLatestContributionsResponse {
repeated ContributionRecord records = 2;
}
// ============== 留言相关消息 ==============
message ActivityMessage {
int64 id = 1;
int64 activity_id = 2;
int64 user_id = 3;
int64 star_id = 4;
string nickname = 5;
string avatar_url = 6;
string content = 7;
int64 created_at = 8;
}
message ListActivityMessagesRequest {
int64 activity_id = 1;
int32 page = 2;
int32 page_size = 3;
}
message ListActivityMessagesResponse {
topfans.common.BaseResponse base = 1;
repeated ActivityMessage messages = 2;
int32 page = 3;
int32 page_size = 4;
int32 total = 5;
}
message CreateActivityMessageRequest {
int64 activity_id = 1;
int64 user_id = 2;
int64 star_id = 3;
string content = 4;
}
message CreateActivityMessageResponse {
topfans.common.BaseResponse base = 1;
ActivityMessage message = 2;
}
// ==================== 活动服务 ====================
service ActivityService {
@ -298,4 +335,19 @@ service ActivityService {
get: "/api/v1/activities/{activity_id}/contributions/latest"
};
}
// 列出活动留言
rpc ListActivityMessages(ListActivityMessagesRequest) returns (ListActivityMessagesResponse) {
option (google.api.http) = {
get: "/api/v1/activities/{activity_id}/messages"
};
}
// 发送一条留言
rpc CreateActivityMessage(CreateActivityMessageRequest) returns (CreateActivityMessageResponse) {
option (google.api.http) = {
post: "/api/v1/activities/{activity_id}/messages"
body: "*"
};
}
}

61
backend/services/activityService/config/config.go Normal file
View File

@ -0,0 +1,61 @@
package config
import (
"os"
"strconv"
"strings"
)
// 默认敏感词列表（首版本地词表，后续接 dify）
var defaultBannedWords = []string{
"傻逼", "操你", "草泥马", "fuck", "shit",
}
// ActivityMessageConfig 活动留言配置
type ActivityMessageConfig struct {
MessageRateLimitPerMin int64 // 单用户单活动每分钟最多留言数
MessageLimitPerActivity int64 // 单用户单活动累计留言上限
BannedWords []string // 敏感词首版本地词表（可通过 env 覆盖）
}
// LoadMessageConfig 从环境变量加载配置（缺省值参考通知服务）
func LoadMessageConfig() *ActivityMessageConfig {
return &ActivityMessageConfig{
MessageRateLimitPerMin: getEnvInt64("ACTIVITY_MESSAGE_RATE_LIMIT_PER_MIN", 5),
MessageLimitPerActivity: getEnvInt64("ACTIVITY_MESSAGE_LIMIT_PER_ACTIVITY", 100),
BannedWords: loadBannedWords(),
}
}
// loadBannedWords 从 ACTIVITY_MESSAGE_BANNED_WORDS（逗号分隔）加载敏感词；
// 未设置或为空时回退到默认词表
func loadBannedWords() []string {
raw := os.Getenv("ACTIVITY_MESSAGE_BANNED_WORDS")
if strings.TrimSpace(raw) == "" {
return append([]string{}, defaultBannedWords...)
}
parts := strings.Split(raw, ",")
out := make([]string, 0, len(parts))
for _, p := range parts {
t := strings.TrimSpace(p)
if t != "" {
out = append(out, t)
}
}
if len(out) == 0 {
return append([]string{}, defaultBannedWords...)
}
return out
}
func getEnvInt64(key string, defaultVal int64) int64 {
v := os.Getenv(key)
if v == "" {
return defaultVal
}
parsed, err := strconv.ParseInt(v, 10, 64)
if err != nil {
return defaultVal
}
return parsed
}

1
backend/services/activityService/main.go
View File

@ -150,6 +150,7 @@ func autoMigrate() error {
&models.ActivityItem{},
&models.ActivityContribution{},
&models.ActivityUserStats{},
&models.ActivityMessage{},
&models.MintingActivity{},
}

36
backend/services/activityService/provider/activity_provider.go
View File

@ -279,3 +279,39 @@ func (p *ActivityProvider) GetLatestContributions(ctx context.Context, req *pb.G
return resp, nil
}
// ListActivityMessages 列出活动留言
func (p *ActivityProvider) ListActivityMessages(ctx context.Context, req *pb.ListActivityMessagesRequest) (*pb.ListActivityMessagesResponse, error) {
logger.Logger.Info("Received ListActivityMessages request", zap.Int64("activity_id", req.ActivityId))
resp, err := p.activityService.ListActivityMessages(ctx, req)
if err != nil {
logger.Logger.Error("ListActivityMessages failed", zap.Error(err))
return &pb.ListActivityMessagesResponse{
Base: &pbCommon.BaseResponse{
Code: uint32(appErrors.ToGRPCCode(err)),
Message: err.Error(),
Timestamp: time.Now().UnixMilli(),
},
}, err
}
logger.Logger.Info("ListActivityMessages successful", zap.Int64("activity_id", req.ActivityId), zap.Int("count", len(resp.Messages)))
return resp, nil
}
// CreateActivityMessage 发送一条留言
func (p *ActivityProvider) CreateActivityMessage(ctx context.Context, req *pb.CreateActivityMessageRequest) (*pb.CreateActivityMessageResponse, error) {
logger.Logger.Info("Received CreateActivityMessage request", zap.Int64("activity_id", req.ActivityId), zap.Int64("user_id", req.UserId))
resp, err := p.activityService.CreateActivityMessage(ctx, req)
if err != nil {
logger.Logger.Error("CreateActivityMessage failed", zap.Error(err))
return &pb.CreateActivityMessageResponse{
Base: &pbCommon.BaseResponse{
Code: uint32(appErrors.ToGRPCCode(err)),
Message: err.Error(),
Timestamp: time.Now().UnixMilli(),
},
}, err
}
logger.Logger.Info("CreateActivityMessage successful", zap.Int64("message_id", resp.Message.Id))
return resp, nil
}

111
backend/services/activityService/repository/activity_messages_repository.go Normal file
View File

@ -0,0 +1,111 @@
package repository
import (
"errors"
"time"
"github.com/topfans/backend/pkg/database"
"github.com/topfans/backend/pkg/models"
"gorm.io/gorm"
)
// ActivityMessagesRepository 活动留言仓库接口
type ActivityMessagesRepository interface {
// Insert 插入一条留言，返回新 ID
Insert(msg *models.ActivityMessage) (int64, error)
// ListByActivity 列出活动的留言（分页，按 created_at DESC, id DESC）
ListByActivity(activityID int64, page, pageSize int) ([]*models.ActivityMessage, int64, error)
// CountByUserActivity 统计某用户在某活动的留言数（用于累计上限校验）
CountByUserActivity(activityID, userID int64) (int64, error)
// UpdateProfile 更新留言的昵称头像（写后异步补字段，避免读时RPC反查）
UpdateProfile(msgID int64, nickname, avatarURL string) error
}
// activityMessagesRepository 实现
type activityMessagesRepository struct {
db *gorm.DB
}
// NewActivityMessagesRepository 创建仓库实例
func NewActivityMessagesRepository() ActivityMessagesRepository {
return &activityMessagesRepository{
db: database.GetDB(),
}
}
// Insert 插入一条留言，返回新 ID
func (r *activityMessagesRepository) Insert(msg *models.ActivityMessage) (int64, error) {
if msg == nil {
return 0, errors.New("message cannot be nil")
}
if err := r.db.Create(msg).Error; err != nil {
return 0, err
}
return msg.ID, nil
}
// ListByActivity 列出活动的留言
func (r *activityMessagesRepository) ListByActivity(activityID int64, page, pageSize int) ([]*models.ActivityMessage, int64, error) {
if activityID <= 0 {
return nil, 0, errors.New("activity_id must be greater than 0")
}
if page <= 0 {
page = 1
}
if pageSize <= 0 {
pageSize = 20
}
if pageSize > 50 {
pageSize = 50
}
query := r.db.Model(&models.ActivityMessage{}).
Where("activity_id = ? AND deleted_at IS NULL AND status = 0", activityID)
var total int64
if err := query.Count(&total).Error; err != nil {
return nil, 0, err
}
var messages []*models.ActivityMessage
offset := (page - 1) * pageSize
if err := query.Order("created_at ASC, id ASC").
Offset(offset).
Limit(pageSize).
Find(&messages).Error; err != nil {
return nil, 0, err
}
return messages, total, nil
}
// CountByUserActivity 统计某用户在某活动的留言数
func (r *activityMessagesRepository) CountByUserActivity(activityID, userID int64) (int64, error) {
if activityID <= 0 || userID <= 0 {
return 0, errors.New("activity_id and user_id must be greater than 0")
}
var count int64
if err := r.db.Model(&models.ActivityMessage{}).
Where("activity_id = ? AND user_id = ? AND deleted_at IS NULL", activityID, userID).
Count(&count).Error; err != nil {
return 0, err
}
return count, nil
}
// UpdateProfile 更新留言的昵称头像
func (r *activityMessagesRepository) UpdateProfile(msgID int64, nickname, avatarURL string) error {
if msgID <= 0 {
return errors.New("msg_id must be greater than 0")
}
return r.db.Model(&models.ActivityMessage{}).
Where("id = ?", msgID).
Updates(map[string]interface{}{
"nickname": nickname,
"avatar_url": avatarURL,
"updated_at": time.Now().UnixMilli(),
}).Error
}

367
backend/services/activityService/service/activity_service.go
View File

@ -2,8 +2,11 @@ package service
import (
"context"
"encoding/json"
"fmt"
"strings"
"time"
"unicode/utf8"
appErrors "github.com/topfans/backend/pkg/errors"
"github.com/topfans/backend/pkg/logger"
@ -11,12 +14,25 @@ import (
pb "github.com/topfans/backend/pkg/proto/activity"
pbCommon "github.com/topfans/backend/pkg/proto/common"
"github.com/topfans/backend/services/activityService/client"
"github.com/topfans/backend/services/activityService/config"
"github.com/topfans/backend/services/activityService/repository"
"github.com/redis/go-redis/v9"
"go.uber.org/zap"
"google.golang.org/grpc/codes"
)
// rateLimitScript 原子 INCR + 首次 EXPIRE 的 Lua 脚本
// 消除 INCR 与 EXPIRE 之间的非原子竞态
var rateLimitScript = func() *redis.Script {
return redis.NewScript(`
local count = redis.call('INCR', KEYS[1])
if count == 1 then
redis.call('EXPIRE', KEYS[1], ARGV[1])
end
return count
`)
}()
// ActivityService 活动Service接口
type ActivityService interface {
// GetActivityList 获取活动列表
@ -45,14 +61,22 @@ type ActivityService interface {
// GetLatestContributions 获取最新贡献记录（用于实时显示）
GetLatestContributions(ctx context.Context, req *pb.GetLatestContributionsRequest) (*pb.GetLatestContributionsResponse, error)
// CreateActivityMessage 创建一条活动留言
CreateActivityMessage(ctx context.Context, req *pb.CreateActivityMessageRequest) (*pb.CreateActivityMessageResponse, error)
// ListActivityMessages 列出活动留言
ListActivityMessages(ctx context.Context, req *pb.ListActivityMessagesRequest) (*pb.ListActivityMessagesResponse, error)
}
// activityService 活动Service实现
type activityService struct {
activityRepo repository.ActivityRepository
mintingActivityRepo repository.MintingActivityRepository
messagesRepo repository.ActivityMessagesRepository
userRPCClient client.UserRPCClient
redisClient *redis.Client
messageCfg *config.ActivityMessageConfig
}
// NewActivityService 创建活动Service实例
@ -60,8 +84,10 @@ func NewActivityService(activityRepo repository.ActivityRepository, mintingActiv
return &activityService{
activityRepo: activityRepo,
mintingActivityRepo: mintingActivityRepo,
messagesRepo: repository.NewActivityMessagesRepository(),
userRPCClient: userRPCClient,
redisClient: redisClient,
messageCfg: config.LoadMessageConfig(),
}
}
@ -410,6 +436,41 @@ func (s *activityService) PurchaseItem(ctx context.Context, req *pb.PurchaseItem
// 更新 Redis 连击计数器（3秒TTL）
s.incrementComboCount(ctx, userID, req.ItemType)
comboCount := s.getComboCount(ctx, userID, req.ItemType)
// Redis Publish contributions_response（实时推送给订阅者）
if s.redisClient != nil {
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
}
contribMsg := &pb.ContributionRecord{
Id: contribution.ID,
UserId: userID,
Nickname: nickname,
AvatarUrl: avatarURL,
StarId: req.StarId,
ItemId: item.ID,
ItemType: req.ItemType,
ItemName: itemName,
ItemIcon: itemIcon,
Quantity: int32(req.Quantity),
ComboCount: int32(comboCount),
CreatedAt: contribution.CreatedAt,
}
payload, _ := json.Marshal(map[string]interface{}{
"activity_id": req.ActivityId,
"type": "contributions_response",
"record": contribMsg,
})
s.redisClient.Publish(ctx, fmt.Sprintf("act:%d:contributions", req.ActivityId), payload)
}
// 更新用户统计
stats, _ := s.activityRepo.GetUserStats(req.ActivityId, userID, req.StarId)
@ -663,6 +724,36 @@ func (s *activityService) BatchPurchaseItem(ctx context.Context, req *pb.BatchPu
// 更新 Redis 连击计数器（3秒TTL）
s.incrementComboCount(ctx, userID, item.ItemType)
comboCount := s.getComboCount(ctx, userID, item.ItemType)
// Redis Publish contributions_response（实时推送给订阅者）
if s.redisClient != nil {
nickname, avatarURL := "", ""
if profile, _ := s.userRPCClient.GetFanProfile(userID, req.StarId); profile != nil {
nickname = profile.Nickname
avatarURL = profile.AvatarUrl
}
contribMsg := &pb.ContributionRecord{
Id: contribution.ID,
UserId: userID,
Nickname: nickname,
AvatarUrl: avatarURL,
StarId: req.StarId,
ItemId: activityItem.ID,
ItemType: item.ItemType,
ItemName: activityItem.ItemName,
ItemIcon: activityItem.IconURL,
Quantity: int32(item.Quantity),
ComboCount: int32(comboCount),
CreatedAt: contribution.CreatedAt,
}
payload, _ := json.Marshal(map[string]interface{}{
"activity_id": req.ActivityId,
"type": "contributions_response",
"record": contribMsg,
})
s.redisClient.Publish(ctx, fmt.Sprintf("act:%d:contributions", req.ActivityId), payload)
}
}
// 更新用户统计
@ -1011,3 +1102,279 @@ func (s *activityService) GetLatestContributions(ctx context.Context, req *pb.Ge
Records: records,
}, nil
}
// CreateActivityMessage 创建一条活动留言（含频控/累计上限/敏感词校验 + Redis Publish）
func (s *activityService) CreateActivityMessage(ctx context.Context, req *pb.CreateActivityMessageRequest) (*pb.CreateActivityMessageResponse, error) {
logger.Logger.Info("CreateActivityMessage request",
zap.Int64("activity_id", req.ActivityId),
zap.Int64("user_id", req.UserId),
zap.Int64("star_id", req.StarId),
)
// 1. 入参校验
// 业务校验错误统一返回 (resp, nil): 把 gRPC code 放进 BaseResponse,
// 让 controller 通过 ErrorWithCode 走 HTTP 200 + body.code 路径;
// 仅基础设施错误(DB/Redis/RPC)用 (nil, err) 让 controller 走 HTTP 500。
content := strings.TrimSpace(req.Content)
if content == "" {
return &pb.CreateActivityMessageResponse{
Base: &pbCommon.BaseResponse{
Code: uint32(appErrors.ToGRPCCode(appErrors.ErrActivityMessageContentEmpty)),
Message: appErrors.ErrActivityMessageContentEmpty.Error(),
},
}, nil
}
if utf8.RuneCountInString(content) > 500 {
return &pb.CreateActivityMessageResponse{
Base: &pbCommon.BaseResponse{
Code: uint32(appErrors.ToGRPCCode(appErrors.ErrActivityMessageContentTooLong)),
Message: appErrors.ErrActivityMessageContentTooLong.Error(),
},
}, nil
}
// 2. 活动存在性 + 状态
activity, err := s.activityRepo.GetActivityByID(req.ActivityId)
if err != nil {
logger.Logger.Error("GetActivityByID failed", zap.Error(err))
return nil, err
}
if activity == nil {
return &pb.CreateActivityMessageResponse{
Base: &pbCommon.BaseResponse{
Code: uint32(appErrors.ToGRPCCode(appErrors.ErrActivityNotFound)),
Message: appErrors.ErrActivityNotFound.Error(),
},
}, nil
}
// 使用 GetCurrentStatus() 计算动态状态(基于 StartTime/EndTime/CurrentProgress),
// 而不是 activity.Status(DB 列,创建后默认值 "pending" 不会被自动更新为 "active")
if activity.GetCurrentStatus() != "active" {
return &pb.CreateActivityMessageResponse{
Base: &pbCommon.BaseResponse{
Code: uint32(appErrors.ToGRPCCode(appErrors.ErrActivityMessageActivityInactive)),
Message: appErrors.ErrActivityMessageActivityInactive.Error(),
},
}, nil
}
// 3. 频控（Redis Lua 原子化：INCR + 首次设置 EXPIRE 60s）
if s.redisClient != nil {
rateKey := fmt.Sprintf("msg:rate:%d:%d", req.ActivityId, req.UserId)
var count int64
if rateLimitScript != nil {
raw, err := rateLimitScript.Run(ctx, s.redisClient, []string{rateKey}, 60).Result()
if err == nil {
if v, ok := raw.(int64); ok {
count = v
}
}
} else {
// 降级：脚本未初始化时退回 INCR + EXPIRE（非原子，但能跑）
count, err = s.redisClient.Incr(ctx, rateKey).Result()
if err == nil && count == 1 {
s.redisClient.Expire(ctx, rateKey, 60*time.Second)
}
}
if count > s.messageCfg.MessageRateLimitPerMin {
return &pb.CreateActivityMessageResponse{
Base: &pbCommon.BaseResponse{
Code: uint32(appErrors.ToGRPCCode(appErrors.ErrActivityMessageTooFrequent)),
Message: appErrors.ErrActivityMessageTooFrequent.Error(),
},
}, nil
}
}
// 4. 累计上限
total, err := s.messagesRepo.CountByUserActivity(req.ActivityId, req.UserId)
if err != nil {
logger.Logger.Error("CountByUserActivity failed", zap.Error(err))
return nil, err
}
if total >= s.messageCfg.MessageLimitPerActivity {
return &pb.CreateActivityMessageResponse{
Base: &pbCommon.BaseResponse{
Code: uint32(appErrors.ToGRPCCode(appErrors.ErrActivityMessageLimitReached)),
Message: appErrors.ErrActivityMessageLimitReached.Error(),
},
}, nil
}
// 5. 敏感词校验（首版本地词表）
if containsBannedWord(content, s.messageCfg.BannedWords) {
return &pb.CreateActivityMessageResponse{
Base: &pbCommon.BaseResponse{
Code: uint32(appErrors.ToGRPCCode(appErrors.ErrActivityMessageContentInvalid)),
Message: appErrors.ErrActivityMessageContentInvalid.Error(),
},
}, nil
}
// 6. 写入
now := time.Now().UnixMilli()
msg := &models.ActivityMessage{
ActivityID: req.ActivityId,
UserID: req.UserId,
StarID: req.StarId,
Nickname: "",
AvatarURL: "",
Content: content,
Status: 0,
CreatedAt: now,
UpdatedAt: now,
}
msgID, err := s.messagesRepo.Insert(msg)
if err != nil {
logger.Logger.Error("Insert message failed", zap.Error(err))
return nil, err
}
// 7. 回查昵称头像
nickname, avatarURL := "", ""
if profile, _ := s.userRPCClient.GetFanProfile(req.UserId, req.StarId); profile != nil {
nickname = profile.Nickname
avatarURL = profile.AvatarUrl
}
// 写回 nickname/avatar（如果首次 RPC 拿到）
if (nickname != "" || avatarURL != "") && msgID > 0 {
if err := s.messagesRepo.UpdateProfile(msgID, nickname, avatarURL); err != nil {
logger.Logger.Warn("UpdateProfile failed", zap.Error(err), zap.Int64("msg_id", msgID))
}
}
pbMsg := &pb.ActivityMessage{
Id: msgID,
ActivityId: req.ActivityId,
UserId: req.UserId,
StarId: req.StarId,
Nickname: nickname,
AvatarUrl: avatarURL,
Content: content,
CreatedAt: now,
}
// 8. Redis Publish（不论是否有人订阅都发）
if s.redisClient != nil {
channel := fmt.Sprintf("act:%d:messages", req.ActivityId)
payload, _ := json.Marshal(map[string]interface{}{
"activity_id": req.ActivityId,
"type": "messages_response",
"message": pbMsg,
})
s.redisClient.Publish(ctx, channel, payload)
}
logger.Logger.Info("CreateActivityMessage success",
zap.Int64("msg_id", msgID),
zap.Int64("user_id", req.UserId),
)
return &pb.CreateActivityMessageResponse{
Base: &pbCommon.BaseResponse{
Code: uint32(codes.OK),
Message: "ok",
},
Message: pbMsg,
}, nil
}
// ListActivityMessages 列出活动留言
func (s *activityService) ListActivityMessages(ctx context.Context, req *pb.ListActivityMessagesRequest) (*pb.ListActivityMessagesResponse, error) {
logger.Logger.Info("ListActivityMessages request",
zap.Int64("activity_id", req.ActivityId),
zap.Int32("page", req.Page),
zap.Int32("page_size", req.PageSize),
)
page := int(req.Page)
pageSize := int(req.PageSize)
if page <= 0 {
page = 1
}
if pageSize <= 0 {
pageSize = 20
}
if pageSize > 50 {
pageSize = 50
}
// 校验活动存在
activity, err := s.activityRepo.GetActivityByID(req.ActivityId)
if err != nil {
logger.Logger.Error("GetActivityByID failed", zap.Error(err))
return nil, err
}
if activity == nil {
return &pb.ListActivityMessagesResponse{
Base: &pbCommon.BaseResponse{
Code: uint32(appErrors.ToGRPCCode(appErrors.ErrActivityNotFound)),
Message: appErrors.ErrActivityNotFound.Error(),
},
}, appErrors.ErrActivityNotFound
}
rows, total, err := s.messagesRepo.ListByActivity(req.ActivityId, page, pageSize)
if err != nil {
logger.Logger.Error("ListByActivity failed", zap.Error(err))
return nil, err
}
messages := make([]*pb.ActivityMessage, 0, len(rows))
for _, m := range rows {
nickname := m.Nickname
avatarURL := m.AvatarURL
// 若 DB 中 nickname 为空（早期数据），回查 user profile
if nickname == "" && avatarURL == "" {
nickname, avatarURL = s.fetchUserProfile(m.UserID, m.StarID)
}
messages = append(messages, &pb.ActivityMessage{
Id: m.ID,
ActivityId: m.ActivityID,
UserId: m.UserID,
StarId: m.StarID,
Nickname: nickname,
AvatarUrl: avatarURL,
Content: m.Content,
CreatedAt: m.CreatedAt,
})
}
logger.Logger.Debug("ListActivityMessages success",
zap.Int64("activity_id", req.ActivityId),
zap.Int("count", len(messages)),
zap.Int64("total", total),
)
return &pb.ListActivityMessagesResponse{
Base: &pbCommon.BaseResponse{
Code: uint32(codes.OK),
Message: "ok",
},
Messages: messages,
Page: int32(page),
PageSize: int32(pageSize),
Total: int32(total),
}, nil
}
// fetchUserProfile 拉取用户昵称头像（失败时返回空串）
func (s *activityService) fetchUserProfile(userID, starID int64) (string, string) {
profile, err := s.userRPCClient.GetFanProfile(userID, starID)
if err != nil || profile == nil {
return "", ""
}
return profile.Nickname, profile.AvatarUrl
}
// containsBannedWord 检查是否含敏感词
func containsBannedWord(content string, banned []string) bool {
lower := strings.ToLower(content)
for _, w := range banned {
if strings.Contains(lower, strings.ToLower(w)) {
return true
}
}
return false
}

2125
docs/superpowers/plans/2026-06-22-activity-realtime-websocket.md Normal file
View File

File diff suppressed because it is too large Load Diff

138
frontend/pages/support-activity/components/ContributionList.vue
View File

@ -2,12 +2,21 @@
<view class="contribution-list" v-if="visible">
<!-- 渐变模糊背景层 -->
<view class="list-bg"></view>
<scroll-view class="list-content" scroll-y>
<scroll-view
class="list-content"
scroll-y
:scroll-into-view="latestRecordId"
:scroll-with-animation="true"
:show-scrollbar="false"
:enhanced="true"
:bounces="false"
>
<view
v-for="(record, index) in records"
:key="record.id"
:id="`contribution-record-${record.id}`"
class="contribution-item"
:class="{ 'new-item': index === 0, 'fading-out': record.fading }"
:class="{ 'new-item': index === records.length - 1, 'fading-out': record.fading }"
>
<!-- 左侧：用户信息（头像 + 昵称 + 赠送动作） -->
<view class="user-info">
@ -44,8 +53,8 @@
</template>
<script setup>
import { ref, computed, onMounted, onUnmounted } from "vue";
import { useContributionPolling } from "../composables/useContributionPolling.js";
import { ref, computed, watch, onMounted, onUnmounted } from "vue";
import { useContributionRealtime } from "../composables/useContributionRealtime.js";
const props = defineProps({
activityId: {
@ -56,46 +65,28 @@ const props = defineProps({
const isPageActive = ref(true);
// 使用轮询 composable
const { records, visible, loading, error, start, stop, reset } =
useContributionPolling(
computed(() => props.activityId),
isPageActive,
);
// 使用实时 composable（WS 优先，断线降级为轮询）
const { records, visible, loading, error, isUsingWS } = useContributionRealtime(
computed(() => props.activityId),
isPageActive,
);
onMounted(() => {
isPageActive.value = true;
});
// 页面生命周期集成
// 实时 composable 内部已在 onMounted/onUnmounted 中处理 start/stop/reset，
// 此处无需重复调用 stop()。
onUnmounted(() => {
stop();
// 仅将页面标记为非激活；实际资源释放由 useContributionRealtime 内部完成。
isPageActive.value = false;
});
// 暴露给父组件使用
defineExpose({
reset,
// useContributionRealtime 内部管理生命周期，无需对外暴露 reset。
});
/**
* 估算单行文本宽度（rpx）
* - CJK 字符按 1.0 * fontSize 计算（方形字宽 ≈ fontSize）
* - 其它字符按 0.55 * fontSize 计算（拉丁/数字字宽 ≈ 半个 fontSize）
*/
const estimateTextWidth = (text, fontSize) => {
if (!text) return 0;
let width = 0;
for (const ch of String(text)) {
// CJK Unified Ideographs + 标点 + 日韩文范围
if (/[　-鿿＀-￯]/.test(ch)) {
width += fontSize * 1.0;
} else {
width += fontSize * 0.55;
}
}
return Math.round(width);
};
/**
* 根据数量值映射数字字号档位
* - >= 999 → count-largest（最大）
@ -109,31 +100,74 @@ const getCountSizeClass = (value) => {
return "count-small";
};
/**
* 根据昵称（+赠送道具名）动态计算胶囊宽度
* 组成：左右内边距(32) + 头像(78) + 头像间距(16) + 文字宽度 + 文字右侧间距(16)
* + 数量区宽度（X + 数字，数字按 1 位约 56rpx 兜底）
*/
const getItemWidth = (record) => {
const nickname = record?.nickname || "";
const giftName = record?.item_name || "";
// scroll-into-view 直接滚到指定 id 的元素,比 scroll-top 更可靠(不用猜位置)。
// 通过"先清空、再设值"的方式,确保每次都能触发 scroll-view 滚动
// (因为 scroll-into-view 值未变化时不会重新滚动)。
//
// 关键设计:不能依赖 records[0] 或 records[N-1] 来定位"最新记录"——
// WS 路径是 append(最新在尾),polling 路径是 prepend(最新在头),方向相反。
// 由于 id 单调递增(后端保证 + useContributionRealtime 也用 highestIdRef 判断),
// 所以直接找 id 最大的那条记录,无论它在头还是在尾,都能正确滚过去。
const latestRecordId = ref("");
let scrollTimerId = null;
const nicknameWidth = estimateTextWidth(nickname, 24);
const giftWidth = estimateTextWidth("送" + giftName, 20);
// 文字列宽度取两者中较宽的一行
const textWidth = Math.max(nicknameWidth, giftWidth);
function getMaxIdRecord() {
const list = records.value;
if (list.length === 0) return null;
let max = list[0];
for (let i = 1; i < list.length; i++) {
if (list[i].id > max.id) max = list[i];
}
return max;
}
// 数量区：item-x (32) + 间距(6) + item-count（按 1 位 56rpx 兜底，多位由 font-size 调整但宽度更窄）
// const quantityWidth = 32 + 6 + 56;
function scrollToLatest() {
// 防抖:快速连续触发时只滚最后一次
if (scrollTimerId) clearTimeout(scrollTimerId);
scrollTimerId = setTimeout(() => {
const target = getMaxIdRecord();
if (!target) return;
const targetId = `contribution-record-${target.id}`;
// 关键:清空再赋值,确保 scroll-into-view 一定触发
latestRecordId.value = "";
// 用 setTimeout 把赋值推到下一个宏任务,避开 Vue 的批处理
setTimeout(() => {
latestRecordId.value = targetId;
}, 0);
scrollTimerId = null;
}, 50);
}
// 固定部分：左内边距16 + 头像78 + 头像右侧间距16 + 文字右侧间距16 + 右内边距16
const FIXED = 16 + 78 + 16 + 16 + 16;
// 监听"id 最大的那条记录的 id"——这是真正能感知"新记录到达"的信号。
// 注意:不能只 watch length —— 当列表已满 MAX_RECORDS=5 时,
// WS 是 [...records, record].slice(-5),polling 也是 [...newRecords, ...records].slice(0, 5),
// 两种情况下 length 都不变,只会替换现有元素,length watch 不触发。
watch(
() => {
const target = getMaxIdRecord();
return target ? target.id : null;
},
() => scrollToLatest(),
);
const total = FIXED + textWidth;
// 监听 visible 变化:scroll-view 通过 v-if="visible" 渲染,visible 由 false→true
// 时 scroll-view 才挂载,需要在挂载完成后立即滚动一次
watch(
() => visible.value,
(val) => {
if (val) {
scrollToLatest();
}
},
);
// 组件挂载后,如果已有历史记录,也立即滚动到 id 最大的那条
onMounted(() => {
if (records.value.length > 0) {
scrollToLatest();
}
});
// 兜底 min/max 与 CSS 保持一致
return Math.min(420, Math.max(200, total));
};
</script>
<style scoped>

41
frontend/pages/support-activity/components/MessageBoard.vue
View File

@ -1,6 +1,14 @@
<template>
<view class="message-board">
<scroll-view class="message-board-scroll" scroll-y :scroll-top="scrollTop">
<scroll-view
class="message-board-scroll"
scroll-y
:scroll-top="scrollTop"
:scroll-with-animation="true"
:show-scrollbar="false"
:enhanced="true"
:bounces="false"
>
<view v-for="msg in messages" :key="msg.id" class="message-bubble">
<text class="msg-user">{{ msg.user }}：</text>
<text class="msg-content">{{ msg.content }}</text>
@ -13,7 +21,7 @@
</template>
<script setup>
import { ref, watch, nextTick } from "vue";
import { ref, watch, nextTick, onMounted } from "vue";
const props = defineProps({
messages: {
@ -22,16 +30,32 @@ const props = defineProps({
},
});
const scrollTop = ref(0);
// scroll-top 需要 string 类型,每次都用一个递增的"超大值"作为目标位置,
// 既保证 scroll-view 一定滚动到底部,又因为值唯一(Vue 响应式触发)
// 即使连续推多条消息也能逐条滚动。
const scrollTop = ref("0");
const scrollVersion = ref(0);
// 监听消息变化,自动滚动到底部
function scrollToBottom() {
scrollVersion.value += 1;
nextTick(() => {
// 递增到比内容高度大得多的值,确保始终滚到最底部
scrollTop.value = String(99999 + scrollVersion.value * 1000);
});
}
// 监听消息数量变化(新增/删除),自动滚动到底部
watch(
() => props.messages.length,
async () => {
await nextTick();
scrollTop.value = 99999;
},
() => scrollToBottom(),
);
// 组件挂载后,如果已有历史消息,也立即滚动到底部
onMounted(() => {
if (props.messages.length > 0) {
scrollToBottom();
}
});
</script>
<style scoped>
@ -64,6 +88,7 @@ watch(
transform: rotate(-0.44deg) skewX(-0.1deg);
animation: bubbleFadeIn 0.3s ease-out;
box-sizing: border-box;
margin-bottom: 4rpx;
}
.msg-user {

59
frontend/pages/support-activity/components/MessageInput.vue
View File

@ -9,7 +9,21 @@
<view class="message-row__pill">
<!-- 内嵌凹槽 (238x30, radius 16) -->
<view class="message-row__groove">
<text class="message-row__text">{{ displayText }}</text>
<!--
真·输入框:输入框默认空,placeholder 复用 displayText 展示"轮播祝福语"。
用户开始键入 → placeholder 自动消失(native placeholder 行为),
emoji 切换只更新 placeholder,不会覆盖用户已经输入的内容。
-->
<input
class="message-row__input"
:value="inputValue"
:placeholder="displayText"
placeholder-class="message-row__input-placeholder"
confirm-type="send"
:maxlength="200"
@input="onInput"
@confirm="handleSend"
/>
</view>
</view>
@ -122,6 +136,20 @@ const displayText = computed(() => {
return "今天第一天姐妹们大家冲鸭！！！";
});
// 输入框本地值:默认空串,只受用户键入控制;
// 父组件的 props.text(emoji 切换 / 直接赋值)不再回灌进 inputValue,
// 而是作为 placeholder(displayText)展示 —— 这样:
// 1) 输入框永远是空的,placeholder 一键消失(native placeholder 行为),用户不用手动删
// 2) emoji 切换不会覆盖用户已经输入的内容,只切换 placeholder
const inputValue = ref("");
function onInput(e) {
// uniapp 小程序 / h5: e.detail.value;某些端直接传字符串;两种都兼容
const val = typeof e === "string" ? e : e?.detail?.value ?? "";
inputValue.value = val;
emit("update:text", val);
}
function handleTap() {
emit("tap");
}
@ -133,8 +161,16 @@ function handleEmoji() {
}
function handleSend() {
// 展示型组件:点击发送图标抛出事件,业务侧决定弹出输入面板 / 直接发送等行为
emit("send", displayText.value);
// 兜底语义:用户未输入时发送 displayText(占位文案);有输入时优先发输入内容
const text = inputValue.value || displayText.value;
emit("send", text);
// 发送后清空输入框:
// 1) 本地 inputValue 清空 → 下次点 send 才会真正走到 fallback 分支
// 2) emit update:text("") → 父组件 messageDraft 同步清空 → displayText 重新计算
// → placeholder 重新可见(native placeholder 行为)
// 否则 inputValue 一直残留旧值,后续点击 send 永远发的都是上一次的文字,fallback 不会触发
inputValue.value = "";
emit("update:text", "");
}
</script>
@ -184,23 +220,32 @@ function handleSend() {
overflow: hidden;
}
.message-row__text {
.message-row__input {
display: block;
width: 100%;
height: 60rpx; /* 30px,撑满凹槽高度 */
font-size: 22rpx; /* 11px */
line-height: 1.3;
color: #ffffff;
font-weight: bold;
font-family: "Abhaya Libre ExtraBold", "PingFang SC", sans-serif;
text-shadow: -2rpx 2rpx 8rpx rgba(206, 9, 9, 0.45);
white-space: nowrap;
overflow: hidden;
text-overflow: ellipsis;
background: transparent;
border: none;
outline: none;
padding: 0;
/* 设计稿的轻微倾斜 */
transform: rotate(-0.44deg) skewX(-0.1deg);
transform-origin: center;
}
.message-row__input-placeholder {
color: rgba(255, 255, 255, 0.9);
font-weight: bold;
font-family: "Abhaya Libre ExtraBold", "PingFang SC", sans-serif;
text-shadow: -2rpx 2rpx 8rpx rgba(206, 9, 9, 0.45);
}
/* 左侧头像 (突出 pill 左缘) */
.message-row__avatar {
position: absolute;

14
frontend/pages/support-activity/composables/useContributionPolling.js
View File

@ -72,7 +72,8 @@ export function useContributionPolling(activityId, isPageActive, options = {}) {
const newRecords = res.data?.records || []
if (newRecords.length === 0) return
// API 返回的 records 按 created_at DESC, id DESC 排序。
// API 返回正序（ASC），用 reverse 反转成"新→旧"，便于复用 [0] 作最新游标 + 前插
newRecords.reverse()
// 只需比对第一条；后面的记录时间戳/ID 必然不更小。
const firstRecord = newRecords[0]
const isNew = firstRecord.created_at > latestTimestamp ||
@ -81,7 +82,7 @@ export function useContributionPolling(activityId, isPageActive, options = {}) {
if (isNew) {
// 重置所有现有记录的计时器（新数据到来，刷新列表）
records.value.forEach(resetRecordTimer)
// 一次性插入本轮拉取到的全部新记录（API 已按"新→旧"排序，新的在前）
// 一次性插入本轮拉取到的全部新记录（reverse 后新→旧，新的在前）
records.value = [...newRecords, ...records.value].slice(0, MAX_RECORDS)
// 为每条新记录启动消失计时器
newRecords.forEach(resetRecordTimer)
@ -110,6 +111,8 @@ export function useContributionPolling(activityId, isPageActive, options = {}) {
}
const newRecords = res.data?.records || []
// API 返回正序（ASC），reverse 反转后 UI 列表"最新在最前"
newRecords.reverse()
// 清除所有计时器
recordTimers.forEach(timer => clearTimeout(timer))
@ -121,7 +124,7 @@ export function useContributionPolling(activityId, isPageActive, options = {}) {
// 为每条记录启动计时器
newRecords.forEach(record => resetRecordTimer(record))
// 更新 latestTimestamp 和 latestId
// 更新 latestTimestamp 和 latestId（reverse 后 [0] 即最新）
if (newRecords.length > 0) {
latestTimestamp = newRecords[0].created_at
latestId = newRecords[0].id
@ -202,6 +205,9 @@ export function useContributionPolling(activityId, isPageActive, options = {}) {
error,
start,
stop,
reset
reset,
// 暴露给 useContributionRealtime 用
highestTimestampRef: () => latestTimestamp,
highestIdRef: () => latestId,
}
}

89
frontend/pages/support-activity/composables/useContributionRealtime.js Normal file
View File

@ -0,0 +1,89 @@
import { onMounted, onUnmounted } from 'vue'
import { useContributionPolling } from './useContributionPolling.js'
import { getActivitySocket } from '@/utils/socket/ActivitySocket.js'
/**
* 贡献实时推送 composable（WS 优先，断线降级为轮询）
* @param {import('vue').Ref<string|number>} activityId
* @param {import('vue').Ref<boolean>} isPageActive
*/
export function useContributionRealtime(activityId, isPageActive) {
const MAX_RECORDS = 5
const {
records,
visible,
loading,
error,
start: startPolling,
stop: stopPolling,
reset: resetPolling,
highestIdRef,
} = useContributionPolling(activityId, isPageActive)
const socket = getActivitySocket()
let usingWS = false
function onWsMessage(payload) {
if (!payload || Number(payload.activity_id) !== Number(activityId.value)) return
if (!payload.record) return
const record = payload.record
if (record.id > highestIdRef()) {
// 追加到列表末尾（与轮询的方向不同），保留最近 MAX_RECORDS 条
records.value = [...records.value, record].slice(-MAX_RECORDS)
}
}
function onWsConnect() {
if (usingWS) return
usingWS = true
stopPolling() // 停掉可能的轮询
socket.subscribe(activityId.value, ['contributions'])
}
function onWsDisconnect() {
if (!usingWS) return
usingWS = false
startPolling() // 降级为轮询
}
socket.onContributionsResponse(onWsMessage)
socket.on('connect', onWsConnect)
socket.on('disconnect', onWsDisconnect)
onMounted(() => {
// 总是调用 connect():SocketManager.connect() 内部会判断 token 是否变化
// (详见 useMessageRealtime.js 注释)。这样能确保用户切换登录后,
// contribution channel 也能用新 token 重新订阅,而不是复用上一个用户的 WS。
const token = uni.getStorageSync('access_token')
if (token) {
socket.connect(token).catch(err => console.warn('[useContributionRealtime] connect error:', err))
}
// 同步分支:如果 WS 已连接(单例复用导致 'connect' 事件不会再次触发),
// 必须直接调 onWsConnect 停轮询,否则 polling 会一直跑。
// 异步分支:WS 还没连上时,先起 polling 兜底;
// 等 'connect' 事件触发 onWsConnect 后会停掉轮询。
if (socket.isConnected) {
onWsConnect()
} else {
startPolling()
}
})
onUnmounted(() => {
if (usingWS) socket.unsubscribe(activityId.value, ['contributions'])
socket.off('connect', onWsConnect)
socket.off('disconnect', onWsDisconnect)
socket.offContributionsResponse(onWsMessage)
stopPolling()
resetPolling()
})
return {
records,
visible,
loading,
error,
isUsingWS: () => usingWS,
}
}

113
frontend/pages/support-activity/composables/useMessageRealtime.js Normal file
View File

@ -0,0 +1,113 @@
import { ref, computed, onMounted, onUnmounted } from 'vue'
import { useStore } from 'vuex'
import { listActivityMessagesApi, createActivityMessageApi } from '@/utils/api.js'
import { getActivitySocket } from '@/utils/socket/ActivitySocket.js'
import { formatRelativeTime } from '@/utils/format.js'
const MAX_MESSAGES = 50
/**
* 活动留言实时推送 composable
* @param {import('vue').Ref<string|number>} activityId
*/
export function useMessageRealtime(activityId) {
const store = useStore()
const messages = ref([])
const loading = ref(false)
const error = ref(null)
const currentUserId = computed(() => store.state.user?.userInfo?.uid)
const socket = getActivitySocket()
// 字段映射：后端 → MessageBoard props
function toComponentShape(m) {
return {
id: m.id,
user: m.nickname || '',
avatar: m.avatar_url || '',
content: m.content,
time: formatRelativeTime(m.created_at),
isSelf: m.user_id === currentUserId.value,
}
}
async function loadHistory() {
if (!activityId.value) return
loading.value = true
error.value = null
try {
const res = await listActivityMessagesApi(activityId.value, 1, 20)
if (res.code === 0) {
messages.value = (res.data.messages || []).map(toComponentShape)
} else {
error.value = res.message || '加载留言失败'
}
} catch (e) {
error.value = e.message || '网络错误'
console.error('[useMessageRealtime] loadHistory error:', e)
} finally {
loading.value = false
}
}
function onWsMessage(payload) {
if (!payload || Number(payload.activity_id) !== Number(activityId.value)) return
if (!payload.message) return
messages.value.push(toComponentShape(payload.message))
if (messages.value.length > MAX_MESSAGES) {
messages.value.splice(0, messages.value.length - MAX_MESSAGES)
}
}
async function sendMessage(content) {
if (!content || !content.trim()) return
const trimmed = content.trim()
try {
const res = await createActivityMessageApi(activityId.value, trimmed)
if (res.code === 0) {
// WS 推回时会再次触发 onMessage → 列表会出现该留言
// 断线时本地 fallback 插入
if (!socket.isConnected && res.data?.message) {
messages.value.push(toComponentShape(res.data.message))
if (messages.value.length > MAX_MESSAGES) {
messages.value.splice(0, messages.value.length - MAX_MESSAGES)
}
}
// 无论 WS 还是 fallback，都弹一个轻量 toast 提示
uni.showToast({ title: '留言成功', icon: 'success', duration: 1200 })
} else {
uni.showToast({ title: res.message || '留言失败', icon: 'none' })
}
} catch (e) {
console.error('[useMessageRealtime] sendMessage error:', e)
uni.showToast({ title: '网络错误', icon: 'none' })
}
}
onMounted(() => {
loadHistory()
// 总是调用 connect():SocketManager.connect() 内部会判断 token 是否变化。
// - 同 token + 已连接 → _doConnect 空跑,无副作用
// - 异 token(用户切换登录) → _hardReset 关闭旧连接、强制重连
// 这样即使旧用户 WS 还活着但 token 已变,也能拿到新用户自己的连接,
// 解决"另一个账户没有接收到实时推送"的问题。
const token = uni.getStorageSync('access_token')
if (token) {
socket.connect(token).catch(err => console.warn('[useMessageRealtime] connect error:', err))
}
socket.subscribe(activityId.value, ['messages'])
socket.onMessagesResponse(onWsMessage)
})
onUnmounted(() => {
socket.unsubscribe(activityId.value, ['messages'])
socket.offMessagesResponse(onWsMessage)
})
return {
messages,
loading,
error,
sendMessage,
refresh: loadHistory,
}
}

61
frontend/pages/support-activity/index.vue
View File

@ -208,6 +208,8 @@ import TopRanking from "./components/TopRanking.vue";
import MessageBoard from "./components/MessageBoard.vue";
import MessageInput from "./components/MessageInput.vue";
import { getEarningsSummaryApi } from "@/utils/api.js";
import { useMessageRealtime } from "./composables/useMessageRealtime.js";
import { useContributionRealtime } from "./composables/useContributionRealtime.js";
const activityType = ref("birthday");
const activityId = ref("");
@ -240,63 +242,16 @@ const currentActivityTitle = ref("");
// 点击骰子 emoji 时按顺序循环选用内置祝福语填入这里
const messageDraft = ref("今天第一天姐妹们大家冲鸭！！！");
// 留言板 mock 数据
const messageList = ref([
{
id: 1,
user: "小星星",
avatar: "",
content: "生日快乐！永远支持你~",
time: "刚刚",
isSelf: false,
},
{
id: 2,
user: "月光宝盒",
avatar: "",
content: "太棒了,继续加油!期待更多精彩作品!",
time: "5分钟前",
isSelf: false,
},
{
id: 3,
user: "彩虹糖",
avatar: "",
content: "为你打call~",
time: "10分钟前",
isSelf: false,
},
{
id: 4,
user: "夜空中最亮的星",
avatar: "",
content: "星光不问赶路人,岁月不负有心人,冲冲冲!",
time: "30分钟前",
isSelf: false,
},
]);
// 留言列表 - 走 useMessageRealtime（HTTP 历史 + WS 实时）
const { messages: messageList, sendMessage } = useMessageRealtime(activityId);
function toggleActionBar() {
actionBarVisible.value = !actionBarVisible.value;
}
// 发送留言
function handleSendMessage(text) {
const newMessage = {
id: Date.now(),
user: "我",
avatar: "",
content: text,
time: "刚刚",
isSelf: true,
};
messageList.value.push(newMessage);
uni.showToast({
title: "留言成功",
icon: "success",
duration: 1500,
});
async function handleSendMessage(text) {
await sendMessage(text);
}
const goBack = () => {
@ -564,8 +519,8 @@ async function initializePage(options = {}) {
// 记录首屏加载时间
performanceMonitor.recordFirstScreenLoad();
initProgressManager();
startProgressSync();
// initProgressManager();
// startProgressSync();
isLoading.value = false;

17
frontend/utils/api.js
View File

@ -874,6 +874,23 @@ export function getActivityContributionsLatestApi(activityId, sinceTimestamp = 0
})
}
// 列出活动留言（首次/下拉加载）
export function listActivityMessagesApi(activityId, page = 1, pageSize = 20) {
return request({
url: `/api/v1/activities/${activityId}/messages?page=${page}&page_size=${pageSize}`,
method: 'GET'
})
}
// 发送一条活动留言
export function createActivityMessageApi(activityId, content) {
return request({
url: `/api/v1/activities/${activityId}/messages`,
method: 'POST',
data: { content }
})
}
// 获取活动贡献点排名
export function getActivityRankingApi(activityId, starId = null, page = 1, pageSize = 10) {
let url = `/api/v1/activities/${activityId}/ranking?page=${page}&page_size=${pageSize}`

40
frontend/utils/format.js Normal file
View File

@ -0,0 +1,40 @@
/**
* 时间格式化工具
*/
/**
* 将毫秒时间戳格式化为相对时间字符串
* - < 60s → "刚刚"
* - < 60min → "X 分钟前"
* - < 24h → "X 小时前"
* - < 7day → "X 天前"
* - >= 7day → 格式化为 YYYY-MM-DD
*
* @param {number} ms 毫秒时间戳
* @returns {string} 相对时间字符串
*/
export function formatRelativeTime(ms) {
if (!ms || typeof ms !== 'number' || ms <= 0) return ''
const now = Date.now()
const diff = Math.max(0, now - ms)
const sec = Math.floor(diff / 1000)
if (sec < 60) return '刚刚'
const min = Math.floor(sec / 60)
if (min < 60) return `${min} 分钟前`
const hr = Math.floor(min / 60)
if (hr < 24) return `${hr} 小时前`
const day = Math.floor(hr / 24)
if (day < 7) return `${day} 天前`
// 7 天及以上显示日期
const d = new Date(ms)
const y = d.getFullYear()
const m = String(d.getMonth() + 1).padStart(2, '0')
const dd = String(d.getDate()).padStart(2, '0')
return `${y}-${m}-${dd}`
}

214
frontend/utils/socket/ActivitySocket.js Normal file
View File

@ -0,0 +1,214 @@
import SocketManager from './SocketManager'
// 活动实时推送 WebSocket 路径
// 优先取 VITE_WS_ACTIVITY_PATH（方便 Nginx 反向代理时统一改前缀），
// 未配置时回退到 /activity
const ACTIVITY_WS_PATH = (() => {
const configured = import.meta.env.VITE_WS_ACTIVITY_PATH
if (configured && String(configured).trim()) {
const p = String(configured).trim()
// 兜底：若用户忘了写前导 /，补上
return p.startsWith('/') ? p : `/${p}`
}
return '/activity'
})()
/**
* 活动实时推送 WebSocket 客户端
* - 继承 SocketManager，复用连接/心跳/重连
* - 额外暴露 topic 订阅 API：subscribe / unsubscribe
*/
// 指数退避：前 5 次按 1s, 2s, 4s, 8s, 16s 重试；之后固定 30s 持续重试（不停止）
const ACTIVITY_RECONNECT_BACKOFF = [1000, 2000, 4000, 8000, 16000]
const ACTIVITY_RECONNECT_FIXED_INTERVAL = 30000
class ActivitySocket extends SocketManager {
constructor(options = {}) {
super({
serviceName: 'Activity',
path: options.path || ACTIVITY_WS_PATH,
// 退避由子类覆写 _tryReconnect 实现；父类仅需一个可用的回退值
reconnectInterval: options.reconnectInterval || 1000,
heartbeatInterval: options.heartbeatInterval || 30000,
maxReconnectAttempts: options.maxReconnectAttempts || Infinity
})
// 退避调度：前 5 次按数组执行，之后固定 30s
this._backoffSchedule = ACTIVITY_RECONNECT_BACKOFF.slice()
this._fixedInterval = ACTIVITY_RECONNECT_FIXED_INTERVAL
// 记录已订阅 topic，便于重连后自动重订阅
this._topics = new Set()
// 特定消息类型回调
this.onMessagesResponseCallback = null
this.onContributionsResponseCallback = null
this._registerActivityHandlers()
}
/**
* 重连：指数退避
* - 前 N 次（N = backoffSchedule.length）：按 1s, 2s, 4s, 8s, 16s 退避
* - 之后：以 30s 固定间隔持续重试（不停止），弱网下也能恢复
*/
_tryReconnect() {
if (this.isClosing) {
console.log(`[${this.serviceName}] Closing intentionally, skip reconnect`)
return
}
if (this.reconnectAttempts >= this.maxReconnectAttempts) {
console.warn(`[${this.serviceName}] Max reconnect attempts reached`)
this._emit('error', { code: 'RECONNECT_FAILED', message: '重连次数已达上限' })
return
}
this.reconnectAttempts++
const idx = this.reconnectAttempts - 1
const delay = idx < this._backoffSchedule.length
? this._backoffSchedule[idx]
: this._fixedInterval
const phase = idx < this._backoffSchedule.length ? 'backoff' : 'fixed'
console.log(
`[${this.serviceName}] Auto reconnecting... ` +
`(attempt ${this.reconnectAttempts}, phase=${phase}, delay=${delay}ms)`
)
this.reconnectTimer = setTimeout(() => {
this._doConnect()
}, delay)
}
_registerActivityHandlers() {
// 留言响应
this.registerHandler('messages_response', (data) => {
if (this.onMessagesResponseCallback) {
this.onMessagesResponseCallback(data)
}
})
// 贡献响应
this.registerHandler('contributions_response', (data) => {
if (this.onContributionsResponseCallback) {
this.onContributionsResponseCallback(data)
}
})
}
/**
* 连接到活动推送服务
*/
async connect(token) {
await super.connect(token, ACTIVITY_WS_PATH)
}
/**
* 订阅活动主题
* @param {string|number} activityId
* @param {string[]} topics - ['messages', 'contributions']
*/
subscribe(activityId, topics = []) {
if (!activityId || !Array.isArray(topics) || topics.length === 0) return
const id = Number(activityId)
topics.forEach(t => this._topics.add(`${id}:${t}`))
if (this.isConnected) {
this.send({ action: 'subscribe', activity_id: id, topics })
}
}
/**
* 取消订阅活动主题
*/
unsubscribe(activityId, topics = []) {
if (!activityId || !Array.isArray(topics) || topics.length === 0) return
const id = Number(activityId)
topics.forEach(t => this._topics.delete(`${id}:${t}`))
if (this.isConnected) {
this.send({ action: 'unsubscribe', activity_id: id, topics })
}
}
/**
* 注册 messages 响应回调
*/
onMessagesResponse(cb) {
this.onMessagesResponseCallback = cb
this.registerHandler('messages_response', (data) => {
if (this.onMessagesResponseCallback) {
this.onMessagesResponseCallback(data)
}
})
}
offMessagesResponse() {
this.onMessagesResponseCallback = null
this.registerHandler('messages_response', () => {})
}
/**
* 注册 contributions 响应回调
*/
onContributionsResponse(cb) {
this.onContributionsResponseCallback = cb
this.registerHandler('contributions_response', (data) => {
if (this.onContributionsResponseCallback) {
this.onContributionsResponseCallback(data)
}
})
}
offContributionsResponse() {
this.onContributionsResponseCallback = null
this.registerHandler('contributions_response', () => {})
}
/**
* 重连成功后，自动重新订阅所有 topic
*/
resubscribeAll() {
if (this._topics.size === 0) return
const byActivity = new Map()
this._topics.forEach(key => {
const [activityId, topic] = key.split(':')
const id = Number(activityId)
if (!byActivity.has(id)) byActivity.set(id, [])
byActivity.get(id).push(topic)
})
byActivity.forEach((topics, activityId) => {
this.send({ action: 'subscribe', activity_id: activityId, topics })
})
}
/**
* SocketManager.connect() 检测到 token 变化时会调用。
* 清空本地 topic 缓存,避免上一个用户的订阅残留导致新用户订阅错乱。
* callbacks(onMessagesResponseCallback / onContributionsResponseCallback)保留,
* 下次组件挂载时会重新设置。
*/
_resetSubscriptions() {
this._topics = new Set()
console.log(`[${this.serviceName}] _resetSubscriptions: cleared topic cache`)
}
}
// ==================== 单例 ====================
let activityInstance = null
export function getActivitySocket() {
if (!activityInstance) {
activityInstance = new ActivitySocket()
// 重连成功后自动重订阅
activityInstance.on('connect', () => activityInstance.resubscribeAll())
}
return activityInstance
}
export function closeActivitySocket() {
if (activityInstance) {
activityInstance.close()
}
}
export default ActivitySocket

12
frontend/utils/socket/GlobalSocketManager.js
View File

@ -1,4 +1,5 @@
import { getAiChatSocket } from './AiChatSocket'
import { getActivitySocket } from './ActivitySocket'
/**
* 全局 WebSocket 管理器
@ -17,6 +18,7 @@ class GlobalSocketManager {
init(token) {
this.token = token
this._initAiChat()
this._initActivity()
// Future: this._initNotification()
}
@ -28,6 +30,16 @@ class GlobalSocketManager {
this.sockets['ai_chat'] = aiChat
}
async _initActivity() {
const activity = getActivitySocket()
activity.on('connect', () => console.log('Activity socket connected'))
activity.on('error', (err) => console.error('Activity socket error:', err))
if (this.token) {
await activity.connect(this.token)
}
this.sockets['activity'] = activity
}
getSocket(serviceName) {
return this.sockets[serviceName]
}

54
frontend/utils/socket/SocketManager.js
View File

@ -38,8 +38,16 @@ class SocketManager {
/**
* 连接到 WebSocket 服务器
* - 若已有 token 且与新 token 不一致：强制关闭旧连接、清空状态、让子类清理订阅缓存，
* 避免上一个用户(以旧 token 鉴权)的 WS 被新用户复用,导致"另一个账户收不到实时推送"。
*/
async connect(token, path) {
const tokenChanged = !!this.token && this.token !== token
if (tokenChanged) {
console.log(`[${this.serviceName}] Token changed, force reconnecting (old token len=${this.token.length}, new token len=${token.length})`)
this._hardReset()
}
this.token = token
this.path = path
this.reconnectAttempts = 0
@ -52,6 +60,38 @@ class SocketManager {
this._doConnect()
}
/**
* 硬重置：关闭现有连接、清空状态。子类可覆写 _resetSubscriptions 清理自己持有的状态(如订阅列表)。
*/
_hardReset() {
// 先标记 isClosing=true 防止 onClose 回调里触发 _tryReconnect
this.isClosing = true
if (this.socket) {
try {
if (typeof this.socket.close === 'function') {
this.socket.close()
} else if (typeof this.socket.complete === 'function') {
this.socket.complete()
}
} catch (err) {
console.warn(`[${this.serviceName}] Close old socket error:`, err)
}
}
// _cleanup 清掉 isConnected/isAuthed/heartbeat/reconnectTimer
this._cleanup()
this.socket = null
this.isClosing = false
// 子类钩子：清掉自己持有的订阅/缓存(如 ActivitySocket._topics)
if (typeof this._resetSubscriptions === 'function') {
try {
this._resetSubscriptions()
} catch (err) {
console.warn(`[${this.serviceName}] _resetSubscriptions error:`, err)
}
}
}
_doConnect() {
// 如果已有连接且已连接，不重复连接
if (this.socket && this.isConnected) {
@ -59,6 +99,13 @@ class SocketManager {
return
}
// 防止多组件在同一 tick 里都调 connect() 导致重复创建 socket
if (this._isConnecting) {
console.log(`[${this.serviceName}] Connect already in progress, skip`)
return
}
this._isConnecting = true
console.log(`[${this.serviceName}] _doConnect called, clearing old socket`)
// 清理旧连接
if (this.socket) {
@ -75,12 +122,14 @@ class SocketManager {
url,
fail: (err) => {
console.error(`[${this.serviceName}] connectSocket fail:`, err)
this._isConnecting = false
this._emit('error', { code: 'CONNECT_FAILED', message: err.errMsg || '连接失败' })
}
})
if (!this.socket) {
console.error(`[${this.serviceName}] socket is null`)
this._isConnecting = false
return
}
@ -88,6 +137,7 @@ class SocketManager {
this._setupListeners()
} catch (err) {
console.error(`[${this.serviceName}] Exception during connect:`, err)
this._isConnecting = false
}
}
@ -108,6 +158,7 @@ class SocketManager {
socket.onOpen(function() {
console.log(`[${self.serviceName}] WebSocket connected`)
self.isConnected = true
self._isConnecting = false // 连接成功,允许后续重连
// 清除重连计时器
if (self.reconnectTimer) {
clearTimeout(self.reconnectTimer)
@ -121,6 +172,7 @@ class SocketManager {
socket.onopen(function() {
console.log(`[${self.serviceName}] WebSocket connected`)
self.isConnected = true
self._isConnecting = false // 连接成功,允许后续重连
// 清除重连计时器
if (self.reconnectTimer) {
clearTimeout(self.reconnectTimer)
@ -150,6 +202,7 @@ class SocketManager {
if (typeof socket.onClose === 'function') {
socket.onClose(function() {
console.log(`[${self.serviceName}] WebSocket closed`)
self._isConnecting = false // 连接关闭,允许后续重连
self._cleanup()
self._emit('disconnect')
self._tryReconnect()
@ -157,6 +210,7 @@ class SocketManager {
} else if (typeof socket.onclose === 'function') {
socket.onclose(function() {
console.log(`[${self.serviceName}] WebSocket closed`)
self._isConnecting = false // 连接关闭,允许后续重连
self._cleanup()
self._emit('disconnect')
self._tryReconnect()

1
frontend/utils/socket/index.js
View File

@ -1,3 +1,4 @@
export { default as SocketManager } from './SocketManager'
export { default as AiChatSocket, getAiChatSocket, closeAiChatSocket, isAiChatClosing, resetAiChatClosing } from './AiChatSocket'
export { default as ActivitySocket, getActivitySocket, closeActivitySocket } from './ActivitySocket'
export { default as GlobalSocketManager, getGlobalSocket } from './GlobalSocketManager'