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