mengyastore/mengyastore-backend-go/internal/mq/client.go

package mq

import (
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"log"
	"strings"
	"sync"
	"time"

	amqp "github.com/rabbitmq/amqp091-go"

	"mengyastore-backend/internal/storage"
)

// Client 封装单条 AMQP 连接、发布用 Channel，以及单环境的命名。
type Client struct {
	conn      *amqp.Connection
	pubCh     *amqp.Channel
	exchange  string
	queue     string
	env       string
	amqpURL   string
	mu        sync.Mutex
	closing   sync.Once
	connected bool

	// 用于重连后重启消费协程（与 StartConsumer 注入的一致）
	consumerCtx  context.Context
	consumerSite *storage.SiteStore
}

// New 连接 RabbitMQ 并声明交换机 + 队列 + 绑定（幂等）。
func New(amqpURL, env string) (*Client, error) {
	if amqpURL == "" {
		return nil, fmt.Errorf("empty amqp url")
	}
	conn, err := amqp.DialConfig(amqpURL, amqp.Config{
		Heartbeat: 30 * time.Second,
		Locale:    "en_US",
	})
	if err != nil {
		return nil, fmt.Errorf("amqp dial: %w", err)
	}

	pubCh, err := conn.Channel()
	if err != nil {
		conn.Close()
		return nil, fmt.Errorf("amqp channel: %w", err)
	}

	exchange, queue, err := declareTopology(pubCh, env)
	if err != nil {
		pubCh.Close()
		conn.Close()
		return nil, err
	}

	return &Client{
		conn:      conn,
		pubCh:     pubCh,
		exchange:  exchange,
		queue:     queue,
		env:       sanitizeEnv(env),
		amqpURL:   amqpURL,
		connected: true,
	}, nil
}

func isRecoverableAMQP(err error) bool {
	if err == nil {
		return false
	}
	var e *amqp.Error
	if errors.As(err, &e) {
		// 504 channel/connection not open、320 连接被服务端关闭等，通过重连恢复
		if e.Code == amqp.ChannelError || e.Code == amqp.ConnectionForced {
			return true
		}
		return strings.Contains(strings.ToLower(e.Reason), "not open")
	}
	s := strings.ToLower(err.Error())
	return strings.Contains(s, "channel/connection is not open") ||
		strings.Contains(s, "connection closed") ||
		strings.Contains(s, "use of closed network connection") ||
		strings.Contains(s, "eof")
}

// reconnectLocked 在持有 mu 时调用：关闭旧连接并重新拨号、声明拓扑。
func (c *Client) reconnectLocked() error {
	if !c.connected || c.amqpURL == "" {
		return fmt.Errorf("mq client closed")
	}

	if c.pubCh != nil {
		_ = c.pubCh.Close()
		c.pubCh = nil
	}
	if c.conn != nil && !c.conn.IsClosed() {
		_ = c.conn.Close()
	}
	c.conn = nil

	conn, err := amqp.DialConfig(c.amqpURL, amqp.Config{
		Heartbeat: 30 * time.Second,
		Locale:    "en_US",
	})
	if err != nil {
		return fmt.Errorf("amqp reconnect dial: %w", err)
	}
	pubCh, err := conn.Channel()
	if err != nil {
		conn.Close()
		return fmt.Errorf("amqp reconnect channel: %w", err)
	}
	exchange, queue, err := declareTopology(pubCh, c.env)
	if err != nil {
		pubCh.Close()
		conn.Close()
		return err
	}
	c.conn = conn
	c.pubCh = pubCh
	c.exchange = exchange
	c.queue = queue

	if c.consumerCtx != nil && c.consumerSite != nil && c.consumerCtx.Err() == nil {
		go RunOrderEmailConsumer(c.consumerCtx, c.conn, c.env, c.consumerSite)
		log.Printf("[MQ] consumer restarted after reconnect (queue=%s)", c.queue)
	}
	return nil
}

func (c *Client) passiveQueueLocked() error {
	if c.pubCh == nil || !c.connected {
		return fmt.Errorf("mq client closed")
	}
	_, err := c.pubCh.QueueDeclarePassive(c.queue, true, false, false, false, nil)
	return err
}

// PublishOrderEmail 向 order-email 路由键发布一条持久化 JSON 消息。
func (c *Client) PublishOrderEmail(ctx context.Context, p OrderEmailPayload) error {
	body, err := json.Marshal(p)
	if err != nil {
		return fmt.Errorf("marshal payload: %w", err)
	}

	c.mu.Lock()
	defer c.mu.Unlock()

	if c.pubCh == nil || !c.connected {
		return fmt.Errorf("mq client closed")
	}

	select {
	case <-ctx.Done():
		return ctx.Err()
	default:
	}

	tryPublish := func() error {
		return c.pubCh.PublishWithContext(ctx,
			c.exchange,
			routingKeyOrderEmail,
			false,
			false,
			amqp.Publishing{
				ContentType:  "application/json",
				DeliveryMode: amqp.Persistent,
				Timestamp:    time.Now(),
				Body:         body,
			},
		)
	}

	err = tryPublish()
	if err != nil && isRecoverableAMQP(err) {
		if rerr := c.reconnectLocked(); rerr == nil {
			err = tryPublish()
		} else {
			err = fmt.Errorf("%w (reconnect: %v)", err, rerr)
		}
	}
	if err != nil {
		return fmt.Errorf("publish: %w", err)
	}
	return nil
}

// QueueInspectInfo 返回当前队列深度与消费者数（被动声明查询）。
func (c *Client) QueueInspectInfo() (messages int, consumers int, err error) {
	c.mu.Lock()
	defer c.mu.Unlock()

	try := func() (int, int, error) {
		if c.pubCh == nil || !c.connected {
			return 0, 0, fmt.Errorf("mq client closed")
		}
		q, e := c.pubCh.QueueDeclarePassive(c.queue, true, false, false, false, nil)
		if e != nil {
			return 0, 0, e
		}
		return q.Messages, q.Consumers, nil
	}

	msgs, cons, err := try()
	if err != nil && isRecoverableAMQP(err) {
		if rerr := c.reconnectLocked(); rerr != nil {
			return 0, 0, fmt.Errorf("%w (reconnect: %v)", err, rerr)
		}
		return try()
	}
	return msgs, cons, err
}

// Ping 检查发布 Channel 能否对声明的队列做被动查询（供 /api/health 探活）。
func (c *Client) Ping() error {
	c.mu.Lock()
	defer c.mu.Unlock()

	err := c.passiveQueueLocked()
	if err != nil && isRecoverableAMQP(err) {
		if rerr := c.reconnectLocked(); rerr != nil {
			return fmt.Errorf("%w (reconnect: %v)", err, rerr)
		}
		return c.passiveQueueLocked()
	}
	return err
}

// Env 返回交换机/队列名使用的环境后缀（已规范化）。
func (c *Client) Env() string { return c.env }

// StartConsumer 在 ctx 取消前运行订单邮件消费者（通常在 goroutine 中调用）。
func (c *Client) StartConsumer(ctx context.Context, site *storage.SiteStore) {
	c.mu.Lock()
	c.consumerCtx = ctx
	c.consumerSite = site
	conn := c.conn
	env := c.env
	c.mu.Unlock()
	if conn == nil || site == nil {
		return
	}
	RunOrderEmailConsumer(ctx, conn, env, site)
}

// Close 关闭发布 Channel 与连接。
func (c *Client) Close() {
	c.closing.Do(func() {
		c.mu.Lock()
		defer c.mu.Unlock()
		c.connected = false
		c.consumerCtx = nil
		c.consumerSite = nil
		if c.pubCh != nil {
			_ = c.pubCh.Close()
			c.pubCh = nil
		}
		if c.conn != nil {
			_ = c.conn.Close()
			c.conn = nil
		}
	})
}