邹方成
a7a0f639e1
fix: 修复微信通知字段截断导致的编码错误 feat: 添加有效邀请相关字段和任务中心常量 refactor: 重构一番赏奖品格位逻辑 perf: 优化道具卡列表聚合显示 docs: 更新项目说明文档和API文档 test: 添加字符串截断工具测试
94 lines
2.3 KiB
Go
94 lines
2.3 KiB
Go
package async
|
|
|
|
import (
|
|
"context"
|
|
"encoding/json"
|
|
"fmt"
|
|
"time"
|
|
|
|
"github.com/redis/go-redis/v9"
|
|
)
|
|
|
|
const (
|
|
TaskEventQueueKey = "task_center:events:queue"
|
|
)
|
|
|
|
type EventType string
|
|
|
|
const (
|
|
EventTypeOrderPaid EventType = "order_paid"
|
|
EventTypeInviteSuccess EventType = "invite_success"
|
|
)
|
|
|
|
type TaskEvent struct {
|
|
Type EventType `json:"type"`
|
|
Payload string `json:"payload"`
|
|
CreatedAt int64 `json:"created_at"`
|
|
}
|
|
|
|
type OrderPaidPayload struct {
|
|
UserID int64 `json:"user_id"`
|
|
OrderID int64 `json:"order_id"`
|
|
}
|
|
|
|
type InviteSuccessPayload struct {
|
|
InviterID int64 `json:"inviter_id"`
|
|
InviteeID int64 `json:"invitee_id"`
|
|
}
|
|
|
|
type TaskQueue interface {
|
|
PublishOrderPaid(ctx context.Context, userID, orderID int64) error
|
|
PublishInviteSuccess(ctx context.Context, inviterID, inviteeID int64) error
|
|
Consume(ctx context.Context) (*TaskEvent, error)
|
|
}
|
|
|
|
type redisTaskQueue struct {
|
|
client *redis.Client
|
|
}
|
|
|
|
func NewRedisTaskQueue(client *redis.Client) TaskQueue {
|
|
return &redisTaskQueue{client: client}
|
|
}
|
|
|
|
func (q *redisTaskQueue) PublishOrderPaid(ctx context.Context, userID, orderID int64) error {
|
|
payload, _ := json.Marshal(OrderPaidPayload{UserID: userID, OrderID: orderID})
|
|
event := TaskEvent{
|
|
Type: EventTypeOrderPaid,
|
|
Payload: string(payload),
|
|
CreatedAt: time.Now().Unix(),
|
|
}
|
|
bytes, _ := json.Marshal(event)
|
|
return q.client.LPush(ctx, TaskEventQueueKey, bytes).Err()
|
|
}
|
|
|
|
func (q *redisTaskQueue) PublishInviteSuccess(ctx context.Context, inviterID, inviteeID int64) error {
|
|
payload, _ := json.Marshal(InviteSuccessPayload{InviterID: inviterID, InviteeID: inviteeID})
|
|
event := TaskEvent{
|
|
Type: EventTypeInviteSuccess,
|
|
Payload: string(payload),
|
|
CreatedAt: time.Now().Unix(),
|
|
}
|
|
bytes, _ := json.Marshal(event)
|
|
return q.client.LPush(ctx, TaskEventQueueKey, bytes).Err()
|
|
}
|
|
|
|
func (q *redisTaskQueue) Consume(ctx context.Context) (*TaskEvent, error) {
|
|
// Block for 2 seconds
|
|
result, err := q.client.BRPop(ctx, 2*time.Second, TaskEventQueueKey).Result()
|
|
if err != nil {
|
|
if err == redis.Nil {
|
|
return nil, nil // Timeout, no message
|
|
}
|
|
return nil, err
|
|
}
|
|
if len(result) < 2 {
|
|
return nil, fmt.Errorf("invalid redis result")
|
|
}
|
|
|
|
var event TaskEvent
|
|
if err := json.Unmarshal([]byte(result[1]), &event); err != nil {
|
|
return nil, err
|
|
}
|
|
return &event, nil
|
|
}
|