102 lines
2.3 KiB
Go
102 lines
2.3 KiB
Go
package mq
|
|
|
|
import (
|
|
"context"
|
|
"encoding/json"
|
|
"log"
|
|
|
|
amqp "github.com/rabbitmq/amqp091-go"
|
|
|
|
"mengyastore-backend/internal/email"
|
|
"mengyastore-backend/internal/storage"
|
|
)
|
|
|
|
// RunOrderEmailConsumer runs until ctx is done. Must be started in its own goroutine.
|
|
func RunOrderEmailConsumer(ctx context.Context, conn *amqp.Connection, env string, site *storage.SiteStore) {
|
|
if conn == nil || site == nil {
|
|
return
|
|
}
|
|
|
|
ch, err := conn.Channel()
|
|
if err != nil {
|
|
log.Printf("[MQ] consumer channel: %v", err)
|
|
return
|
|
}
|
|
defer ch.Close()
|
|
|
|
if _, _, err := declareTopology(ch, env); err != nil {
|
|
log.Printf("[MQ] consumer declare topology: %v", err)
|
|
return
|
|
}
|
|
|
|
if err := ch.Qos(1, 0, false); err != nil {
|
|
log.Printf("[MQ] consumer qos: %v", err)
|
|
return
|
|
}
|
|
|
|
queue := QueueName(env)
|
|
const tag = "mengyastore-order-email"
|
|
msgs, err := ch.Consume(queue, tag, false, false, false, false, nil)
|
|
if err != nil {
|
|
log.Printf("[MQ] consume %s: %v", queue, err)
|
|
return
|
|
}
|
|
|
|
log.Printf("[MQ] consumer started queue=%s", queue)
|
|
|
|
for {
|
|
select {
|
|
case <-ctx.Done():
|
|
_ = ch.Cancel(tag, false)
|
|
log.Printf("[MQ] consumer stopped queue=%s", queue)
|
|
return
|
|
case d, ok := <-msgs:
|
|
if !ok {
|
|
return
|
|
}
|
|
handleDelivery(site, d)
|
|
}
|
|
}
|
|
}
|
|
|
|
func handleDelivery(site *storage.SiteStore, d amqp.Delivery) {
|
|
var payload OrderEmailPayload
|
|
if err := json.Unmarshal(d.Body, &payload); err != nil {
|
|
log.Printf("[MQ] bad message: %v", err)
|
|
_ = d.Ack(false)
|
|
return
|
|
}
|
|
if payload.ToEmail == "" {
|
|
_ = d.Ack(false)
|
|
return
|
|
}
|
|
|
|
cfg, err := site.GetSMTPConfig()
|
|
if err != nil || !cfg.IsConfiguredEmail() {
|
|
log.Printf("[MQ] skip email order=%s: smtp not configured", payload.OrderID)
|
|
_ = d.Ack(false)
|
|
return
|
|
}
|
|
|
|
emailCfg := email.Config{
|
|
SMTPHost: cfg.Host,
|
|
SMTPPort: cfg.Port,
|
|
From: cfg.Email,
|
|
Password: cfg.Password,
|
|
FromName: cfg.FromName,
|
|
}
|
|
data := payload.ToNotifyData()
|
|
if err := email.SendOrderNotify(emailCfg, data); err != nil {
|
|
log.Printf("[MQ] send email fail order=%s: %v", payload.OrderID, err)
|
|
if d.Redelivered {
|
|
log.Printf("[MQ] drop order=%s after failed redelivery", payload.OrderID)
|
|
_ = d.Ack(false)
|
|
return
|
|
}
|
|
_ = d.Nack(false, true)
|
|
return
|
|
}
|
|
log.Printf("[MQ] email ok order=%s to=%s", payload.OrderID, payload.ToEmail)
|
|
_ = d.Ack(false)
|
|
}
|