Merge remote-tracking branch 'origin/master'

consume/egg_canal_invite_user_nums_consume.go

@@ -59,6 +59,7 @@ func handleEggCanalInviteUserNumsConsume(msg []byte) error {
 	var canalMsg *md.CanalUserRelateMessage[md.CanalUserRelate]
 	err := json.Unmarshal(msg, &canalMsg)
 	if err != nil {
+		fmt.Println("EggCanalInviteUserNumsConsumeFaliedUnMarshal_ERR:::::", err.Error())
 		return nil
 	}
 
@@ -70,19 +71,22 @@ func handleEggCanalInviteUserNumsConsume(msg []byte) error {
 	if canalMsg.Type == md2.CanalMsgInsertSqlType {
 		for _, item := range canalMsg.Data {
 			parentUid := item.ParentUid
-			id := fmt.Sprintf("%d%d-%d", year, week, parentUid)
+			id := fmt.Sprintf("%d%d_%s", year, week, parentUid)
 
 			// 新增拉新人数
 			script := elastic.NewScript("ctx._source.invite_user_nums += params.inc").Param("inc", 1)
-			_, err = es.EsClient.Update().
+			service, err := es.EsClient.Update().
 				Index(index).
 				Id(id).
 				Script(script).
 				Do(context.Background())
 			if err != nil {
+				fmt.Println("EggCanalInviteUserNumsConsumeFailedUpdateInviteUserNums_ERR:::::", err.Error())
 				return err
 			}
+			fmt.Println(service)
 		}
 	}
+	fmt.Println("EggCanalInviteUserNumsConsume_Succeed::::")
 	return nil
 }

consume/egg_canal_person_add_activity_value_consume.go

@@ -80,23 +80,27 @@ func handleEggCanalPersonAddActivityValueConsume(msg []byte) error {
 	if canalMsg.Type == md2.CanalMsgInsertSqlType {
 		for _, item := range canalMsg.Data {
 			uid := item.Uid
-			id := fmt.Sprintf("%d%d-%d", year, week, uid)
+			id := fmt.Sprintf("%d%d_%s", year, week, uid)
 
-			if item.CoinId != personEggPointsCoinId && item.Direction != 1 {
+			if item.CoinId != utils2.IntToStr(personEggPointsCoinId) && item.Direction != "1" {
 				continue
 			}
 
 			// 3. 增加个人活跃积分
-			script := elastic.NewScript("ctx._source.person_add_activity_value += params.inc").Param("inc", item.Amount)
-			_, err = es.EsClient.Update().
+			amount := utils2.StrToFloat64(item.Amount)
+			script := elastic.NewScript("ctx._source.person_add_activity_value += params.inc").Param("inc", int(amount))
+			updateDoc, err := es.EsClient.Update().
 				Index(index).
 				Id(id).
 				Script(script).
 				Do(context.Background())
 			if err != nil {
+				fmt.Println("EggCanalPersonAddActivityValueConsumeUpdateDoc_ERR::::", err.Error())
 				return err
 			}
+			fmt.Println("updateDoc==========>", updateDoc)
 		}
 	}
+	fmt.Println("EggCanalPersonAddActivityValueConsume_End::::::")
 	return nil
 }

consume/egg_canal_violate_nums_consume.go

@@ -85,7 +85,7 @@ func handleEggCanalViolateNumsConsume(msg []byte) error {
 				continue
 			}
 			uid := item.Uid
-			id := fmt.Sprintf("%d%d-%d", year, week, uid)
+			id := fmt.Sprintf("%d%d_%s", year, week, uid)
 
 			// 2.2. 增加违规次数记录
 			script := elastic.NewScript("ctx._source.violate_nums += params.inc").Param("inc", 1)
@@ -115,7 +115,7 @@ func handleEggCanalViolateNumsConsume(msg []byte) error {
 				continue
 			}
 			uid := item.Uid
-			id := fmt.Sprintf("%d%d-%d", year, week, uid)
+			id := fmt.Sprintf("%d%d_%s", year, week, uid)
 
 			// 3.2 减少违规次数记录
 			script := elastic.NewScript("ctx._source.violate_nums -= params.dec").Param("dec", 1)

consume/egg_energy_deal_user_ecpm.go

@@ -67,12 +67,11 @@ func handleEggEnergyDealUserECPMConsume(msgData []byte) error {
 	yearStr := utils2.IntToStr(year)
 	weekStr := utils2.IntToStr(week)
 	index := es2.GetAppointIndexFromAlias(yearStr, weekStr)
-	id := fmt.Sprintf("%d%d-%d", year, week, msg.Uid)
+	id := fmt.Sprintf("%d%d_%d", year, week, msg.Uid)
 
-	m := md2.EggEnergyUserEggScoreEs{
-		Ecpm: utils2.StrToFloat64(msg.Ecpm),
-	}
-	updateDocRet, err := es.UpdateDoc(index, id, m)
+	updateMap := make(map[string]interface{})
+	updateMap["ecpm"] = utils2.StrToFloat64(msg.Ecpm)
+	updateDocRet, err := es.UpdateDoc(index, id, updateMap)
 	if err != nil {
 		return err
 	}

consume/md/consume_key.go

@@ -169,7 +169,7 @@ var RabbitMqQueueKeyList = []*MqQueue{
 		Name:           "egg_canal_invite_user_nums_queue",
 		Type:           DirectQueueType,
 		IsPersistent:   false,
-		RoutKey:        "egg_canal_invite_user_nums",
+		RoutKey:        "user_relate",
 		BindKey:        "",
 		ConsumeFunName: "EggCanalInviteUserNumsConsume",
 	},
@@ -178,7 +178,7 @@ var RabbitMqQueueKeyList = []*MqQueue{
 		Name:           "egg_canal_violate_nums_queue",
 		Type:           DirectQueueType,
 		IsPersistent:   false,
-		RoutKey:        "egg_canal_violate_nums",
+		RoutKey:        "user_tag_records",
 		BindKey:        "",
 		ConsumeFunName: "EggCanalViolateNumsConsume",
 	},
@@ -205,7 +205,7 @@ var RabbitMqQueueKeyList = []*MqQueue{
 		Name:           "egg_canal_person_add_activity_value_queue",
 		Type:           DirectQueueType,
 		IsPersistent:   false,
-		RoutKey:        "egg_canal_person_add_activity_value",
+		RoutKey:        "egg_canal_user_virtual_coin_flow",
 		BindKey:        "",
 		ConsumeFunName: "EggCanalPersonAddActivityValueConsume",
 	},

consume/md/md_egg_canal_person_add_activity_value_consume.go

@@ -1,16 +1,16 @@
 package md
 
 type CanalUserVirtualCoinFlow struct {
-	Id           int64  `json:"id"`
-	Uid          int64  `json:"uid"`
-	CoinId       int    `json:"coin_id"`
-	Direction    int    `json:"direction"`
+	Id           string `json:"id"`
+	Uid          string `json:"uid"`
+	CoinId       string `json:"coin_id"`
+	Direction    string `json:"direction"`
 	Title        string `json:"title"`
 	Amount       string `json:"amount"`
 	BeforeAmount string `json:"before_amount"`
 	AfterAmount  string `json:"after_amount"`
 	SysFee       string `json:"sys_fee"`
-	TransferType int    `json:"transfer_type"`
+	TransferType string `json:"transfer_type"`
 }
 
 type CanalUserVirtualCoinFlowMessage[T any] struct {

consume/md/md_egg_canal_user_relate_consume.go

@@ -1,10 +1,10 @@
 package md
 
 type CanalUserRelate struct {
-	Id         int64  `json:"id"`
-	ParentUid  int64  `json:"parent_uid"`
-	Uid        int64  `json:"uid"`
-	Level      int    `json:"level"`
+	Id         string `json:"id"`
+	ParentUid  string `json:"parent_uid"`
+	Uid        string `json:"uid"`
+	Level      string `json:"level"`
 	InviteTime string `json:"invite_time"`
 }
 

consume/md/md_egg_canal_violate_nums_consume.go

@@ -1,9 +1,9 @@
 package md
 
 type CanalTagRecords struct {
-	Id       int    `json:"id" `
-	TagId    int    `json:"tag_id" `
-	Uid      int64  `json:"uid" `
+	Id       string `json:"id" `
+	TagId    string `json:"tag_id" `
+	Uid      string `json:"uid" `
 	Memo     string `json:"memo" `
 	CreateAt string `json:"create_at" `
 	UpdateAt string `json:"update_at" `

go.mod

@@ -8,7 +8,7 @@ go 1.19
 
 require (
 	code.fnuoos.com/EggPlanet/egg_models.git v0.2.1-0.20241212120727-3681308aeb14
-	code.fnuoos.com/EggPlanet/egg_system_rules.git v0.0.4-0.20241212140020-c99f60b4f868
+	code.fnuoos.com/EggPlanet/egg_system_rules.git v0.0.4-0.20241213073654-f37e71ad92ef
 	code.fnuoos.com/go_rely_warehouse/zyos_go_es.git v1.0.1-0.20241118083738-0f22da9ba0be
 	code.fnuoos.com/go_rely_warehouse/zyos_go_mq.git v0.0.5
 	github.com/boombuler/barcode v1.0.1