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"
	"log"
	"os"
)

var templateMap = map[string]map[string]string{
	"test": {
		"hangup":     "0ByvFSV3B3U5-1XyvruRe7AMHAxiwQBFUL50lmoatCY",
		"autofight":  "5rwMjLs6nVQ0SMtS0EAPEpmXZduOdEgSWbIjSNG5T7M",
		"guildgame":  "e6D_r9_5tKNZK0cvM8fjzMXy9z1-aoNYQ9TSvAEuklU",
		"alliance":   "e6D_r9_5tKNZK0cvM8fjzMXy9z1-aoNYQ9TSvAEuklU",
		"warheavens": "e6D_r9_5tKNZK0cvM8fjzMXy9z1-aoNYQ9TSvAEuklU",
	},
	"product": {
		"hangup":     "wAXZWcbt8Hkxmt2iZXR7ieiM1A_Mqt7cFU5ky6dZNMo",
		"autofight":  "HbFIcVV78vbPr1uhe-tVAo8oQKv2jnljvPYRl-ktiFE",
		"guildgame":  "Q6JBEqIPufrfmsZw-Wds2sdLATfJU2zA0ebpMR3W2F4",
		"alliance":   "Q6JBEqIPufrfmsZw-Wds2sdLATfJU2zA0ebpMR3W2F4",
		"warheavens": "Q6JBEqIPufrfmsZw-Wds2sdLATfJU2zA0ebpMR3W2F4",
	},
}

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)
		}
	}()

	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 {
		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.wxHelper.SendWechatSubscribe(subscribeOne.OpenId, templateMap[os.Getenv("CHANNEL")]["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 {
		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.wxHelper.SendWechatSubscribe(subscribeOne.OpenId, templateMap[os.Getenv("CHANNEL")]["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.wxHelper.SendWechatSubscribe(openId, templateMap[os.Getenv("CHANNEL")]["guildgame"], 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.wxHelper.SendWechatSubscribe(openId, templateMap[os.Getenv("CHANNEL")]["alliance"], 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.wxHelper.SendWechatSubscribe(openId, templateMap[os.Getenv("CHANNEL")]["warheavens"], msg)
	}
}