dsbqj_admin/pkg/sender/subscribe.go

package sender

import (
	"dsbqj-admin/model/mongo/subscribe"
	"dsbqj-admin/pkg/helper/wechat"
	"dsbqj-admin/pkg/logger"
	"github.com/kamva/mgm/v3"
	"go.mongodb.org/mongo-driver/bson"
	yaml "gopkg.in/yaml.v2"
	"log"
	"os"
	"strings"
	"sync"
)

const subscribeTemplateConfigFile = "config/subscribe_template.yaml"

var (
	subscribeTemplateOnce    sync.Once
	subscribeTemplateMap     map[string]map[string]string
	subscribeTemplateLoadErr error
)

type SubscribeSend struct {
	DeviceId string
	OpenIds  []string
	Module   string
}

type SubscribeSender struct {
	workerNum int
	queue     chan *SubscribeSend
	wxHelper  *wechat.WechatHelper
}

func NewSubscribeSender(workerNum int) *SubscribeSender {
	return &SubscribeSender{
		workerNum: workerNum,
		queue:     make(chan *SubscribeSend, 1000),
		wxHelper:  wechat.NewWechatHelper(),
	}
}

func (this *SubscribeSender) Send(send *SubscribeSend) {
	select {
	case this.queue <- send:
	default:
		logger.Info("subscribe sender queue full, drop task=%+v", send)
	}
}

func (this *SubscribeSender) Start() {
	for i := 0; i < this.workerNum; i++ {
		go this.worker(i)
	}
}

func (this *SubscribeSender) worker(id int) {
	for task := range this.queue {
		logger.Info("Subscribe worker(%d) do safe send module %s, openid_len %d device id %s", id, task.Module, len(task.OpenIds), task.DeviceId)
		this.safeSend(task)
	}
}

func (this *SubscribeSender) safeSend(send *SubscribeSend) {
	defer func() {
		if r := recover(); r != nil {
			log.Printf("send subscribe panic: %v", r)
			logger.Info("[订阅推送] 发送订阅消息异常, module=%s, device_id=%s, err=%v", send.Module, send.DeviceId, r)
		}
	}()

	switch send.Module {
	case "hangup":
		this.SendHangupSubscribe(send.DeviceId)
	case "autofight":
		this.SendAutoFightSubscribe(send.DeviceId)
	case "guildgame":
		this.SendGuildGameSubscribe(send.OpenIds)
	case "alliance":
		this.SendAllianceSubscribe(send.OpenIds)
	case "warheavens":
		this.SendWarHeavensSubscribe(send.OpenIds)
	default:
		logger.Info("[push] unknown type=%s, openId=%s",
			send.Module, send.DeviceId)
	}
}

func (this *SubscribeSender) SendHangupSubscribe(deviceId string) {
	subscribeOne := new(subscribe.Subscribe)
	err := mgm.Coll(&subscribe.Subscribe{}).First(bson.M{"device_id": deviceId, "modules.hangup.enabled": true}, subscribeOne)
	if err != nil {
		logger.Info("[订阅推送] 查询订阅记录失败, module=%s, device_id=%s, err=%v", "hangup", deviceId, err)
		return
	}

	msg := make(map[string]map[string]string)
	msg["thing1"] = make(map[string]string)
	msg["thing1"]["value"] = "挂机奖励时长已满"
	msg["thing3"] = make(map[string]string)
	msg["thing3"]["value"] = "您的挂机奖励时长已满，请打开游戏领取"

	this.sendWechatSubscribe(subscribeOne.OpenId, "hangup", msg)
}

func (this *SubscribeSender) SendAutoFightSubscribe(deviceId string) {
	subscribeOne := new(subscribe.Subscribe)
	err := mgm.Coll(&subscribe.Subscribe{}).First(bson.M{"device_id": deviceId, "modules.autofight.enabled": true}, subscribeOne)
	if err != nil {
		logger.Info("[订阅推送] 查询订阅记录失败, module=%s, device_id=%s, err=%v", "autofight", deviceId, err)
		return
	}

	msg := make(map[string]map[string]string)
	msg["thing2"] = make(map[string]string)
	msg["thing2"]["value"] = "离线闯关结束，请收取您的离线闯关奖励。"
	msg["thing1"] = make(map[string]string)
	msg["thing1"]["value"] = "离线闯关提醒"

	this.sendWechatSubscribe(subscribeOne.OpenId, "autofight", msg)
}

func (this *SubscribeSender) SendGuildGameSubscribe(openIds []string) {
	msg := make(map[string]map[string]string)
	msg["thing2"] = make(map[string]string)
	msg["thing2"]["value"] = "门派攻防战即将开始，为门派荣誉而战吧。"
	msg["thing4"] = make(map[string]string)
	msg["thing4"]["value"] = "门派攻防战"

	for _, openId := range openIds {
		this.sendWechatSubscribe(openId, "guildgame", msg)
	}
}

func (this *SubscribeSender) SendAllianceSubscribe(openIds []string) {
	msg := make(map[string]map[string]string)
	msg["thing2"] = make(map[string]string)
	msg["thing2"]["value"] = "三界战场即将开始，争夺人间道统。"
	msg["thing4"] = make(map[string]string)
	msg["thing4"]["value"] = "三界争峰"

	for _, openId := range openIds {
		this.sendWechatSubscribe(openId, "alliance", msg)
	}
}

func (this *SubscribeSender) SendWarHeavensSubscribe(openIds []string) {
	msg := make(map[string]map[string]string)
	msg["thing2"] = make(map[string]string)
	msg["thing2"]["value"] = "独战诸仙即将开始！"
	msg["thing4"] = make(map[string]string)
	msg["thing4"]["value"] = "决战诸仙"

	for _, openId := range openIds {
		this.sendWechatSubscribe(openId, "warheavens", msg)
	}
}

func (this *SubscribeSender) sendWechatSubscribe(openId string, module string, msg map[string]map[string]string) {
	templateID := subscribeTemplateID(module)
	if templateID == "" {
		logger.Info("[push] subscribe template id not configured, channel=%s, module=%s", os.Getenv("CHANNEL"), module)
		return
	}

	this.wxHelper.SendWechatSubscribe(openId, templateID, msg)
}

func subscribeTemplateID(module string) string {
	templates, err := loadSubscribeTemplateConfig()
	if err != nil {
		logger.Info("[push] load subscribe template config failed: %v", err)
		return ""
	}

	channel := strings.ToLower(strings.TrimSpace(os.Getenv("CHANNEL")))
	moduleKey := strings.ToLower(strings.TrimSpace(module))
	if channel != "" {
		if templateID := strings.TrimSpace(templates[channel][moduleKey]); templateID != "" {
			return templateID
		}
	}

	return strings.TrimSpace(templates["default"][moduleKey])
}

func loadSubscribeTemplateConfig() (map[string]map[string]string, error) {
	subscribeTemplateOnce.Do(func() {
		data, err := os.ReadFile(subscribeTemplateConfigFile)
		if err != nil {
			subscribeTemplateLoadErr = err
			return
		}
		subscribeTemplateLoadErr = yaml.Unmarshal(data, &subscribeTemplateMap)
	})

	return subscribeTemplateMap, subscribeTemplateLoadErr
}