wip:告警

entity/entity.go

@@ -0,0 +1,5 @@
+package entity
+
+var (
+	SseClients = map[int64]chan string{}
+)

etc/config.yaml

@@ -6,7 +6,9 @@ etcd:
   # 基础服务
   messageKey: "message.rpc" #消息中台rpc
   address:
-    - 192.168.3.204:2379
+    - 172.31.31.203:2379
+    #- 192.168.3.165:2379
+    #- 192.168.3.204:2379
   baseserver:
     # 用户中台配置
     integral:
@@ -14,30 +16,29 @@ etcd:
   # 剑鱼文库
   jydocs:
     userlib:
-        key: jydocs.userlib.rpc
+      key: jydocs.userlib.rpc
     stdlib:
       key: jydocs.stdlib.rpc
-  userCenterKey: "usercenter.rpc" #用户中台rpc
 #消息队列
 nsq:
+  address: 172.20.45.129:4161
   topic: jy_event
-  address: 192.168.3.240:4161
   channel: event
 
 #数据库配置
 databases:
   # redis配置
-  redis: main=192.168.3.149:1712,session=192.168.3.149:1713,newother=192.168.3.149:1712
+  redis: main=172.20.45.129:1712,session=172.20.45.129:1713,other=172.20.45.129:1712,newother=172.20.45.129:1712
   # nsq操作日志库
   mogLog:
-    address: 192.168.3.206:27090
+    address: 172.20.45.129:27002,172.20.45.130:27080
     size: 5
     dbName: qfw
     replSet:
-    userName: admin
-    password: 123456
+    userName: ""
+    password: ""
   mongodb:
-    address: 192.168.3.206:27080
+    address: 172.20.45.129:27002,172.20.45.130:27080
     size: 5
     dbName: qfw
     replSet:
@@ -45,28 +46,27 @@ databases:
     password:
   mysql:
     dbName: jianyu
-    address: 192.168.3.149:3306
-    userName: root
+    address: 172.20.45.129:4000
+    userName: jianyu
     passWord: Topnet123
     maxOpenConns: 5
     maxIdleConns: 5
   tidb:
     dbName: base_service
-    address: 192.168.3.217:4000
+    address: 172.20.45.129:4000
     userName: root
     passWord: "=PDT49#80Z!RVv52_z"
     maxOpenConns: 5
     maxIdleConns: 5
   tidbPoint:
     dbName: base_service
-    address: 192.168.3.217:4000
+    address: 172.20.45.129:4000
     userName: root
     passWord: "=PDT49#80Z!RVv52_z"
     maxOpenConns: 5
     maxIdleConns: 5
-
 rpc:
-  payrpc: 192.168.3.149:8600
+  payrpc: 172.31.31.203:8600
 
 #其他配置
 newUserAward: #新用户注册奖励
@@ -85,6 +85,7 @@ inviteRegister: #分享裂变活动
 
 reportInvited: #年报分享注册奖励
   points: 500
+
 shareOpenDetail: #三级页分享
   openDetail:
     points: 5
@@ -92,26 +93,13 @@ shareOpenDetail: #三级页分享
     points: 5
     cycle: day
 
-newRegister: #新注册送7天超级订阅
-  startTime: 2024-11-01 00:00:00 #活动开始
-  entTime: 2024-11-12 00:00:00 #活动结束
-  supVip: 7
-  message:
-    title: 注册成功，送您七天超级订阅！
-    content: 剑鱼标讯送您七天超级订阅，默认订阅地区为北京。如需要调整订阅标讯地区，请戳此处链接前往订阅设置。
-    msgType: 2
-    appid: 10000
-    pcUrl: /page_workDesktop/work-bench/app/big/big_subscribe?vt=v
-    mobileUrl: /jyapp/vipsubscribe/toSubVipSetPage
-    wxUrl: /front/vipsubscribe/toSubVipSetPage
-
 jyactivity: # 临时活动
   name: 剑鱼超级会员节
   code: membershipDay
   dateRange: #活动时间
-    t1: 2022-05-25 00:00:00 #预热活动
-    t2: 2022-11-12 00:00:00 #活动开始
-    ed: 2022-12-13 00:00:00 #活动结束
+    t1: 2022-11-10 00:00:00 #预热活动
+    t2: 2022-11-28 15:20:00 #活动开始
+    ed: 2022-12-12 14:50:00 #活动结束
   missions: #任务列表
     buysubvip: #购买超级订阅
       points: 800
@@ -128,17 +116,17 @@ jyactivity: # 临时活动
     1500:
       subvip: 30
   dailyBoon: #预热活动每日领取
-    clock: 10:00:00 #每日开抢时间
+    clock: 09:00:00 #每日开抢时间
     reward:
       svip: 7 #赠送超级订阅限制
-      limit: 500 #每日数量限制
+      limit: 50 #每日数量限制
   power: #预热活动参加资格
     switch: true #开关
     free: true #是否是免费用户
     time: 1609430400 #注册时间限制 2021-01-01 00：00：00
 
 #优惠券Rpc查询接口
-lotteryUrl: http://192.168.3.206:8090
+lotteryUrl: http://172.31.31.203:8090
 productCode:
   subscription: 101
   subscriptionWeek: 1011
@@ -146,25 +134,23 @@ productCode:
   subscriptionQuarter: 1013
   subscriptionYear: 1014
 
-# 订单监控
-orderMonitor:
-  dateSpecial: 2022-10-19         # 特殊奖品活动日期 (第111名活动日期)
-  openCron: "0 0 6 * * *"         # 每天6点开启
-  selectCron: "0 */10 7-23 * * *" # 查询订单的表达式  每十分钟查询一次
-  winNumberDaily: 11       # 每天第xx个付款获得奖品1
-  winNumbersSpecial: 111   # 第xxx个付款获得特殊奖品
 
+# 订单监控
+orderMonitor :
+  dateSpecial: 2022-10-24         # 特殊奖品活动日期 (第111名活动日期)
+  openCron: "0 35 10 * * *"         # 每天6点开启
+  selectCron: "0 */10 7-23 * * *"    # 查询订单的表达式   每十分钟查询一次
+  winNumberDaily: 4       # 每天第xx个付款获得奖品
+  winNumbersSpecial: 6   # 第xxx个付款获得特殊奖品
   switch: true  # 是否开启定时任务
-  activityMode: 2 # 1. 双十一模式（活动期间每天第xx名，）  2. 双十二活动模式（活动期间内第xx,xx,xx,xx名）
-
+  activityMode: 2  # 1. 每天第多少名  2. 活动期间内一共第多少名(双十二)
   rules:
-    - winNum: [ 2,12,22,122,1212,2222 ] # 活动期间第XXX,XXX个支付中奖  有序数组  顺序不要乱
+    - winNum: [ 2,4,6,9,13,15 ] # 活动期间第XXX,XXX个支付中奖  有序数组  顺序不要乱
       mold: 3 # 3 免单
       products: [  "VIP订阅","数据流量包" ] # 参加活动的产品
       priceLimit: true  # 是否有实付金额限制
       priceStart: 30000 # 实付金额最低  单位 分
       priceEnd: 100000  # 实付金额最高 单位 分
-taskStartTime: 1698020000 #新手任务开始时间
 NsqLog:
   Name: nsq日志             # 日志名称
   CollName: nsq_logs   # 保存的coll
@@ -172,4 +158,17 @@ NsqLog:
   SPSize: 3            # 数据库并发数据
   BulkSize: 500        # 每批的数量
   TimeAfter: 5000      # 定时保存 毫秒
-  Timeout: 2000       # 缓存通道满时，超时丢弃 毫秒
+  Timeout: 2000       # 缓存通道满时，超时丢弃
+newRegister: #新注册送7天超级订阅
+  startTime: 2024-11-01 00:00:00 #活动开始
+  entTime: 2037-11-30 00:00:00 #活动结束
+  supVip: 7
+  message:
+    title: 注册成功，送您七天超级订阅！
+    content: 剑鱼标讯送您七天超级订阅，默认订阅地区为北京。如需要调整订阅标讯地区，请戳此处链接前往订阅设置。
+    msgType: 2
+    appid: 10000
+    pcUrl: /page_workDesktop/work-bench/app/big/big_subscribe?vt=v
+    mobileUrl: /jyapp/vipsubscribe/toSubVipSetPage
+    wxUrl: /front/vipsubscribe/toSubVipSetPage
+webhookURL: ["https://qyapi.weixin.qq.com/cgi-bin/webhook/send?key=24b0ac60-3a02-441f-842e-9cd3f75d1208"]

handler/activity/lotteryDrawTask.go

@@ -8,6 +8,8 @@ import (
 	"app.yhyue.com/moapp/message/util"
 	"encoding/json"
 	"fmt"
+	"github.com/gogf/gf/v2/frame/g"
+	"github.com/gogf/gf/v2/os/gctx"
 	"time"
 )
 
@@ -23,11 +25,12 @@ type TaskInfo struct {
 	Id                int64  `json:"id"`
 	Name              string `json:"name"`
 	ActiveId          int64  `json:"active_id"`     //活动id
-	Qualification     string `json:"qualification"` //获取抽奖资格次数；
+	Qualification     int    `json:"qualification"` //获取抽奖资格次数；
 	CycleNum          int    `json:"cycle_num"`     //任务周期（每天/每周/每月/活动周期） 可执行任务次数:n次;-1：不限制
 	CycleUnit         int    `json:"cycle_unit"`    //默认0：当天；；1：当周；2：当月；3：活动周期
 	ActivityStartTime string `json:"start_time"`    //活动开始时间
 	ActivityEndTime   string `json:"end_time"`      //活动结束时间
+	ActivityName      string `json:"activity_name"` //活动名称
 }
 
 // 信息主题
@@ -61,7 +64,7 @@ func LotteryDrawTask(msg *model.Message) {
 		return
 	}
 	//任务信息
-	taskInfos := db.Mysql.SelectBySql(fmt.Sprintf("SELECT lti.*,ai.start_time,ai.end_time FROM %s lti LEFT JOIN %s ai ON lti.active_id = ai.id  WHERE lti.active_id  = ? AND  lti.id = ? AND ai.end_time >= NOW() ORDER BY lti.create_date  DESC", tableTaskInfo, tableActivityInfo), msgBody.ActiveId, msgBody.TaskInfoId)
+	taskInfos := db.Mysql.SelectBySql(fmt.Sprintf("SELECT lti.*,ai.start_time,ai.end_time,ai.name AS activity_name FROM %s lti LEFT JOIN %s ai ON lti.active_id = ai.id  WHERE lti.active_id  = ? AND  lti.id = ? AND ai.end_time >= NOW() ORDER BY lti.create_date  DESC", tableTaskInfo, tableActivityInfo), msgBody.ActiveId, msgBody.TaskInfoId)
 	if taskInfos == nil || len(*taskInfos) == 0 {
 		logger.Info(fmt.Sprintf("没有当前需要完成的任务信息：%v", msgBody))
 		return
@@ -144,9 +147,10 @@ func LotteryDrawTask(msg *model.Message) {
 	case 3: //招标采购搜索任务
 	case 4: //购买/续费/升级超级订阅会员
 	case 5: //购买/续费/升级大会员
+	// TODO  完成任务 赠送两次 抽奖机会
 	case 6: //给朋友分享活动
 	}
-	if id := db.Mysql.Insert(tableTaskUser, map[string]interface{}{
+	insertMap := map[string]interface{}{
 		"active_id":   msgBody.ActiveId,
 		"task_id":     msgBody.TaskInfoId,
 		"phone":       msgBody.Phone,
@@ -157,7 +161,16 @@ func LotteryDrawTask(msg *model.Message) {
 		"end_date":    timeRange.EndTime.Format(date.Date_Full_Layout),
 		"update_date": time.Now().Format(date.Date_Full_Layout),
 		"create_date": time.Now().Format(date.Date_Full_Layout),
-	}); id < 0 {
-		logger.Info(fmt.Sprintf("保存任务记录异常：%v", msgBody))
+	}
+	for i := 0; i < taskInfo.Qualification; i++ {
+		if id := db.Mysql.Insert(tableTaskUser, insertMap); id < 0 {
+			logger.Info(fmt.Sprintf("保存任务记录异常：%v", msgBody))
+			// TODO  保存任务记录异常 告警
+			if webhookURL := g.Cfg().MustGet(gctx.New(), "webhookURL").Strings(); len(webhookURL) > 0 {
+				content := fmt.Sprintf(`保存任务记录异常:\n活动名称：%s \n手机号：%s \n职位id：%d \n任务id：%d \n任务名称：%s`,
+					taskInfo.ActivityName, msgBody.Phone, msgBody.PositionId, msgBody.TaskInfoId, taskInfo.Name)
+				util.SendMsgByWXURL(content, webhookURL)
+			}
+		}
 	}
 }

model/sse.go

@@ -0,0 +1,9 @@
+package model
+
+// 消息结构体，用于传递 sender、message 和时间
+type SseMessage struct {
+	Prize string `json:"prize"` //奖品信息 例如：腾讯视频会员周卡
+	User  string `json:"user"`  //中奖人信息 例如：157****0152
+	State int    `json:"state"` //0：当前用户中奖；1：其他用户中奖
+	Time  string `json:"time"`
+}

services/filter/sessionfilter.go

@@ -2,6 +2,7 @@ package filter
 
 import (
 	"net/http"
+	"strings"
 
 	"app.yhyue.com/moapp/jybase/go-xweb/xweb"
 )
@@ -14,7 +15,7 @@ type sessionfilter struct {
 // 继承过滤器方法
 func (l *sessionfilter) Do(w http.ResponseWriter, req *http.Request) bool {
 	session := l.App.SessionManager.Session(req, w)
-	if req.RequestURI == "/jyActivity/getConfig/info" {
+	if req.RequestURI == "/jyActivity/getConfig/info" || strings.Contains(req.RequestURI, "/sse/") {
 		return true
 	}
 	if session.Get("userId") == nil {

services/init.go

@@ -2,6 +2,9 @@ package services
 
 import (
 	"app.yhyue.com/moapp/message/services/activity"
+	"app.yhyue.com/moapp/message/services/sse"
+	"github.com/gogf/gf/v2/os/gcfg"
+	"github.com/gogf/gf/v2/os/gctx"
 	"time"
 
 	"app.yhyue.com/moapp/jybase/go-xweb/httpsession"
@@ -16,7 +19,7 @@ func init() {
 	//开启redissession
 	httpsession.IsRedisSessionStore = true
 	xweb.Config.Profiler = true
-
+	httpsession.Domain = gcfg.Instance().MustGet(gctx.New(), "cookieDomain").String()
 	xweb.RootApp().AppConfig.StaticFileVersion = false
 	xweb.RootApp().AppConfig.CheckXsrf = false
 	xweb.RootApp().AppConfig.EnableHttpCache = false
@@ -30,4 +33,5 @@ func init() {
 	xweb.AddAction(&bidderPlan.Activity{})
 	xweb.AddAction(&task.Task{})
 	xweb.AddAction(&activity.ConfigRouter{})
+	xweb.AddAction(&sse.ServerSentRouter{})
 }

services/sse/sse.go

@@ -0,0 +1,69 @@
+package sse
+
+import (
+	"app.yhyue.com/moapp/jybase/common"
+	"app.yhyue.com/moapp/jybase/go-logger/logger"
+	"app.yhyue.com/moapp/jybase/go-xweb/xweb"
+	"app.yhyue.com/moapp/message/entity"
+	"app.yhyue.com/moapp/message/model"
+	"encoding/json"
+	"fmt"
+	"net/http"
+	"time"
+)
+
+type ServerSentRouter struct {
+	*xweb.Action
+	events xweb.Mapper `xweb:"/sse/events"`
+}
+
+func (s *ServerSentRouter) Events() {
+	defer common.Catch()
+	s.ResponseWriter.Header().Set("Content-Type", "text/event-stream")
+	s.ResponseWriter.Header().Set("Cache-Control", "no-cache")
+	s.ResponseWriter.Header().Set("Connection", "keep-alive")
+	s.ResponseWriter.Header().Set("Access-Control-Allow-Origin", "*")
+
+	sessVal := s.Session().GetMultiple()
+	if userId := common.ObjToString(sessVal["userId"]); userId != "" {
+		positionId := common.Int64All(sessVal["positionId"])
+		clientChan := make(chan string, 1)
+		entity.SseClients[positionId] = clientChan
+		defer func() {
+			delete(entity.SseClients, positionId)
+			close(clientChan)
+		}()
+		//go SendToClient(positionId)
+		for {
+			select {
+			case msg := <-clientChan:
+				logger.Info("msg:", msg, "---------------------------")
+				fmt.Fprintf(s.ResponseWriter, "data: %s\n\n", msg)
+				s.ResponseWriter.(http.Flusher).Flush()
+			case <-s.Request.Context().Done():
+				return
+			}
+		}
+	}
+}
+
+func SendToClient(positionId int64) {
+	clientChan, exists := entity.SseClients[positionId]
+	if !exists {
+		logger.Info("Client not found: %s", positionId)
+		return
+	}
+	for i := 0; i < 10; i++ {
+		message := fmt.Sprintf("这是发送内容 第 %d 条", i)
+		msg := model.SseMessage{User: "剑鱼", Prize: message, Time: time.Now().Format("15:04:05")}
+		msgData, _ := json.Marshal(msg)
+		select {
+		case clientChan <- string(msgData):
+			logger.Info(fmt.Sprintf("Sent to client %s: %s", positionId, message))
+		default:
+			logger.Info(fmt.Sprintf("Channel full for client %s, message dropped: %s", positionId, message))
+			return
+		}
+		time.Sleep(2 * time.Second)
+	}
+}

util/warn.go

@@ -0,0 +1,42 @@
+package util
+
+import (
+	"bytes"
+	"encoding/json"
+	"github.com/gogf/gf/v2/frame/g"
+	"github.com/gogf/gf/v2/os/gctx"
+	"net/http"
+)
+
+func SendMsgByWXURL(msg string, whs []string) {
+	for _, url := range whs {
+		if ok := SendBot(url, msg); !ok {
+			g.Log().Info(gctx.New(), "企业微信机器人提醒失败--：", url, msg)
+		}
+	}
+}
+
+func SendBot(webhookURL, msg string) (b bool) {
+	// 构造请求体
+	payload := map[string]interface{}{
+		"msgtype": "text",
+		"text": map[string]string{
+			"content": msg,
+		},
+	}
+	// 转换为 JSON 字符串
+	payloadBytes, err := json.Marshal(payload)
+	if err != nil {
+		g.Log().Info(gctx.New(), "Error :", err.Error())
+		return
+	}
+	// 发送 POST 请求
+	resp, err := http.Post(webhookURL, "application/json", bytes.NewReader(payloadBytes))
+	if err != nil {
+		g.Log().Info(gctx.New(), "Error :", err.Error())
+		return
+	}
+	defer resp.Body.Close()
+	b = true
+	return
+}