common/message/redis_msg.go

package message

import (
	"context"
	"fmt"
	"strings"
	"time"

	"github.com/gogf/gf/v2/frame/g"
	"github.com/gogf/gf/v2/os/glog"
	"github.com/gogf/gf/v2/util/gconv"
)

type RedisPublishMsgConfig struct {
	QueueName string
	Data      any
}

type RedisSubscribeMsgConfig struct {
	QueueName     string
	ConsumerName  string
	AutoAck       bool
	PrefetchCount int
	HandleFunc    func(ctx context.Context, message map[string]interface{}) error
}

func (*RedisPublishMsgConfig) GetPublishMsgType() {

}

func (*RedisSubscribeMsgConfig) GetSubscribeMsgType() {

}

func init() {
	// 注册 Redis 插件（连接由 RegisterPlugin 异步处理）
	registerPlugin(MessageRedis, func() messageUtil {
		return &redis{}
	})
}

type redis struct{}

// RedisStreamMessage Redis Stream 消息结构
type redisStreamMessage struct {
	ID     string
	Values map[string]interface{}
}

// Ping 检测 Redis 连接状态
func (c *redis) ping(ctx context.Context) bool {
	conn, err := getDefaultDataSource()
	if err != nil {
		return false
	}
	return conn.redisPing(ctx)
}

// Reconnect 重连 Redis
func (c *redis) reconnect(ctx context.Context) error {
	conn, err := getDefaultDataSource()
	if err != nil {
		return fmt.Errorf("获取默认连接失败: %w", err)
	}

	if err := conn.redisReconnect(ctx); err != nil {
		return fmt.Errorf("redis重连失败: %w", err)
	}

	return nil
}

// Close 关闭 Redis 连接
func (c *redis) close(ctx context.Context) error {
	conn, err := getDefaultDataSource()
	if err != nil {
		return fmt.Errorf("获取默认连接失败: %w", err)
	}

	if err := conn.redisClose(ctx); err != nil {
		return fmt.Errorf("关闭redis连接失败: %w", err)
	}

	return nil
}

// Publish 发布消息
func (c *redis) Publish(ctx context.Context, msgConfig messagePublishConfig) error {
	cfg, ok := msgConfig.(*RedisPublishMsgConfig)
	if !ok {
		return fmt.Errorf("无效的 Redis 配置类型")
	}
	if g.IsEmpty(cfg.QueueName) {
		return fmt.Errorf("队列名称不能为空")
	}
	if g.IsEmpty(cfg.Data) {
		return fmt.Errorf("数据不能为空")
	}
	conn, err := getDefaultDataSource()
	if err != nil {
		return fmt.Errorf("获取默认连接失败: %w", err)
	}

	if !conn.getIsConnected() {
		if err := conn.redisReconnect(ctx); err != nil {
			return fmt.Errorf("redis重连失败: %w", err)
		}
	}

	values := gconv.Map(cfg.Data)
	args := make([]interface{}, 0, len(values)*2+2)
	args = append(args, cfg.QueueName, "*")
	for key, val := range values {
		args = append(args, key, val)
	}
	result, err := conn.getClient().Do(ctx, "XADD", args...)
	if err != nil {
		g.Log().Errorf(ctx, "❌ Redis 发布消息失败: key=%s, err=%v", cfg.QueueName, err)
		return err
	}
	g.Log().Infof(ctx, "✅ Redis 发布消息成功: key=%s, messageID=%s", cfg.QueueName, gconv.String(result))
	return nil
}

// Subscribe 订阅消息
func (c *redis) Subscribe(ctx context.Context, msgConfig messageSubscribeConfig) error {
	cfg, ok := msgConfig.(*RedisSubscribeMsgConfig)
	if !ok {
		return fmt.Errorf("无效的 Redis 配置类型")
	}
	if g.IsEmpty(cfg.QueueName) {
		return fmt.Errorf("队列名称不能为空")
	}
	if g.IsEmpty(cfg.ConsumerName) {
		return fmt.Errorf("消费者名称不能为空")
	}
	if g.IsEmpty(cfg.HandleFunc) {
		return fmt.Errorf("处理函数不能为空")
	}
	return c.createSubscribe(ctx, cfg.QueueName, cfg.ConsumerName, cfg.PrefetchCount, cfg.AutoAck, cfg.HandleFunc)
}

// createSubscribe 内部订阅消息
func (c *redis) createSubscribe(ctx context.Context, key, consumerName string, prefetchCount int, autoAck bool, handler func(ctx context.Context, message map[string]interface{}) error) error {
	go func() {
		defer func() {
			if r := recover(); r != nil {
				g.Log().Errorf(ctx, "❌ Redis 消费者 panic: %v", r)
			}
		}()

		retryTicker := time.NewTicker(time.Second)
		defer retryTicker.Stop()

		// 重试计数器
		var consecutiveErrors int
		const maxConsecutiveErrors = 3

		for {
			select {
			case <-ctx.Done():
				g.Log().Infof(ctx, "🔕 Redis 消费者停止: topic=%s", key)
				return
			case <-retryTicker.C:
				err := c.consumeMessages(ctx, key, consumerName, prefetchCount, autoAck, handler)
				if err != nil {
					// 对于超时错误,返回nil继续循环,而不是返回错误
					if strings.Contains(err.Error(), "i/o timeout") || strings.Contains(err.Error(), "timeout") ||
						strings.Contains(err.Error(), "context deadline exceeded") || strings.Contains(err.Error(), "context canceled") {

						consecutiveErrors++
						if consecutiveErrors > maxConsecutiveErrors {
							g.Log().Errorf(ctx, "Max retries exceeded, giving up")
							return
						}
						backoffTime := 5 * time.Second
						g.Log().Warningf(ctx, "⚠️ 等待 %v 后重试...", backoffTime)

						time.Sleep(backoffTime)
					} else {
						// 非超时错误（严重错误）
						consecutiveErrors = 0 // 重置计数
						g.Log().Errorf(ctx, "严重错误，立即重试: %v", err)

						// 短暂等待后重试
						select {
						case <-ctx.Done():
							return
						case <-time.After(time.Second):
							// 继续循环
						}
					}
				} else {
					// 成功时重置错误计数器
					consecutiveErrors = 0
				}
			}
		}
	}()
	return nil
}

// consumeMessages 消费消息
func (c *redis) consumeMessages(ctx context.Context, key, consumerName string, prefetchCount int, autoAck bool, handler func(ctx context.Context, message map[string]interface{}) error) error {
	conn, err := getDefaultDataSource()
	if err != nil {
		return fmt.Errorf("获取默认连接失败: %w", err)
	}

	if !conn.getIsConnected() {
		if err := conn.redisReconnect(ctx); err != nil {
			return fmt.Errorf("redis重连失败: %w", err)
		}
	}

	// 检查消费者组是否存在
	if err := c.createStreamGroup(ctx, key); err != nil {
		return fmt.Errorf("create stream group failed: %w", err)
	}

	// 使用带重试的命令执行
	result, err := conn.getClient().Do(ctx, "XREADGROUP", "GROUP", "default", consumerName, "COUNT", prefetchCount, "BLOCK", 0, "STREAMS", key, ">")
	if err != nil {
		if strings.Contains(err.Error(), "i/o timeout") || strings.Contains(err.Error(), "timeout") ||
			strings.Contains(err.Error(), "context deadline exceeded") || strings.Contains(err.Error(), "context canceled") {

		}
		return err
	}
	messages, err := c.parseStreamResult(result)
	if err != nil {
		return err
	}
	for _, msg := range messages {
		// 处理消息
		if err := handler(ctx, msg.Values); err != nil {
			g.Log().Errorf(ctx, "❌ 消息处理失败: messageID=%s, err=%v", msg.ID, err)
			continue
		}

		// ACK 消息
		if autoAck {
			if err := c.ackMessage(ctx, key, "default", msg.ID); err != nil {
				g.Log().Errorf(ctx, "❌ ACK 消息失败: messageID=%s, err=%v", msg.ID, err)
			}
		}
	}

	return nil
}

// createStreamGroup 内部单个创建消费组
func (c *redis) createStreamGroup(ctx context.Context, key string) error {
	conn, err := getDefaultDataSource()
	if err != nil {
		return fmt.Errorf("获取默认连接失败: %w", err)
	}

	if !conn.getIsConnected() {
		if err := conn.redisReconnect(ctx); err != nil {
			return fmt.Errorf("redis重连失败: %w", err)
		}
	}

	groupName := "default"
	_, err = conn.getClient().Do(ctx, "XGROUP", "CREATE", key, groupName, "0", "MKSTREAM")
	if err != nil {
		errStr := err.Error()
		if strings.Contains(errStr, "BUSYGROUP") && strings.Contains(errStr, "already exists") {
			glog.Infof(ctx, "✅ Redis 消费者组已存在: %s", groupName)
			return nil
		}
		return fmt.Errorf("初始化消费者组失败: %w", err)
	}
	glog.Infof(ctx, "✅ Redis 消费者组创建成功: %s", groupName)
	return nil
}

// ackMessage ACK 消息
func (c *redis) ackMessage(ctx context.Context, streamKey, groupName string, messageIDs ...string) error {
	conn, err := getDefaultDataSource()
	if err != nil {
		return fmt.Errorf("获取默认连接失败: %w", err)
	}

	if !conn.getIsConnected() {
		if err := conn.redisReconnect(ctx); err != nil {
			return fmt.Errorf("redis重连失败: %w", err)
		}
	}

	args := make([]interface{}, 0, len(messageIDs)+2)
	args = append(args, streamKey, groupName)
	for _, id := range messageIDs {
		args = append(args, id)
	}
	_, err = conn.getClient().Do(ctx, "XACK", args...)
	return err
}

// parseStreamResult 解析 Stream 结果
func (c *redis) parseStreamResult(result interface{}) ([]redisStreamMessage, error) {
	if result == nil {
		return []redisStreamMessage{}, nil
	}

	var resultVal interface{}

	// 尝试获取 Val() 方法
	if valuer, ok := result.(interface{ Val() interface{} }); ok {
		resultVal = valuer.Val()
	} else {
		resultVal = result
	}

	// 检查是否为空
	if resultVal == nil {
		return []redisStreamMessage{}, nil
	}

	// 预分配切片容量,避免多次扩容
	messages := make([]redisStreamMessage, 0)

	if streamsMap, ok := resultVal.(map[interface{}]interface{}); ok {
		for _, streamData := range streamsMap {
			msgArray, ok := streamData.([]interface{})
			if !ok {
				continue
			}
			for _, msgData := range msgArray {
				msgArray, ok := msgData.([]interface{})
				if !ok || len(msgArray) < 2 {
					continue
				}
				msgID := gconv.String(msgArray[0])
				fieldsArray, ok := msgArray[1].([]interface{})
				if !ok {
					continue
				}
				values := make(map[string]interface{}, len(fieldsArray)/2)
				for i := 0; i < len(fieldsArray); i += 2 {
					if i+1 < len(fieldsArray) {
						key := gconv.String(fieldsArray[i])
						values[key] = fieldsArray[i+1]
					}
				}
				messages = append(messages, redisStreamMessage{
					ID:     msgID,
					Values: values,
				})
			}
		}
	}

	return messages, nil
}