wechat-api/crontask/send_wx.go

package crontask

import (
	"context"
	"encoding/json"
	"github.com/imroc/req/v3"
	"strings"
	"time"
	"wechat-api/ent"
	"wechat-api/ent/messagerecords"
	"wechat-api/ent/server"
	"wechat-api/ent/wx"
	"wechat-api/internal/types/payload"
)

func (l *CronTask) sendWx() {
	ctx := context.Background()
	config :=
		`{
    "ip": "172.18.41.219",
    "number": 10,
    "wx_list": [
        {
            "port": "30001",
            "wxid": "wxid_77au928zeb2p12",
            "nickname": "百事通"
        },
        {
            "port": "30007",
            "wxid": "wxid_edc0mvp188ms22",
            "nickname": "爱博闻"
        },
        {
            "port": "30040",
            "wxid": "wxid_wupstwe851nb12",
            "nickname": "轻马小助手"
        }
    ]
}`

	var p payload.SendWxPayload
	if err := json.Unmarshal([]byte(config), &p); err != nil {
		l.Logger.Errorf("failed to unmarshal the payload :%v", err)
		return
	}

	// 更新最旧的 p.Number 条 status 值为 1 或 4 的 message_records 记录的 status 为 2
	messages, err := l.svcCtx.DB.MessageRecords.Query().
		Where(messagerecords.StatusIn(1, 4)).
		Order(ent.Asc(messagerecords.FieldCreatedAt)).
		Limit(p.Number).
		All(ctx)
	if err != nil {
		l.Logger.Errorf("get messageRecords list failed %v", err)
		return
	}

	startTime := time.Now()

	client := req.C().DevMode()
	client.SetCommonRetryCount(2).
		SetCommonRetryBackoffInterval(1*time.Second, 5*time.Second).
		SetCommonRetryFixedInterval(2 * time.Second).SetTimeout(30 * time.Second)

	type SendTextMsgReq struct {
		Wxid string `json:"wxid"`
		Msg  string `json:"msg"`
	}

	type SendFileMsgReq struct {
		Wxid        string `json:"wxid"`
		Filepath    string `json:"filepath"`
		Diyfilename string `json:"diyfilename"`
	}

	// 从 Redis 中获取服务器信息(ip, port)
	getServerInfo := func(wxid string) (string, error) {
		key := "crontask_wx_server_info"
		val, err := l.svcCtx.Rds.HGet(ctx, key, wxid).Result()
		if err != nil {
			return "", err
		}

		if val == "" {
			wx, err := l.svcCtx.DB.Wx.Query().Where(wx.WxidEQ(wxid)).First(l.ctx)
			if err != nil {
				l.Logger.Errorf("get wx info failed wxid=%v err=%v", wxid, err)
				return "", err
			}
			server, err := l.svcCtx.DB.Server.Query().Where(server.IDEQ(wx.ServerID)).First(l.ctx)
			if err != nil {
				l.Logger.Errorf("get server info failed wxid=%v err=%v", wxid, err)
				return "", err
			}
			val = server.PrivateIP + ":" + server.AdminPort
			l.svcCtx.Rds.HSet(ctx, key, wxid, val)
		}

		return val, nil
	}

	for _, v := range messages {
		serverInfo, _ := getServerInfo(v.BotWxid)
		infoArray := strings.Split(serverInfo, ":")
		ip, wxPort := infoArray[0], infoArray[1]

		// 更新 status 值为 2（发送中）
		tx, err := l.svcCtx.DB.Tx(l.ctx)

		_, err = tx.MessageRecords.UpdateOneID(v.ID).SetStatus(2).Save(ctx)
		if err != nil {
			l.Logger.Errorf("update messageRecords failed id=%v err=%v", v.ID, err)
			continue
		}

		if v.ContentType == 1 {
			_, err = client.R().SetBody(&SendTextMsgReq{
				Wxid: v.ContactWxid,
				Msg:  v.Content,
			}).Post("http://" + ip + ":" + wxPort + "/SendTextMsg")
		} else {
			_, err = client.R().SetBody(&SendFileMsgReq{
				Wxid:        v.ContactWxid,
				Filepath:    v.Content,
				Diyfilename: v.Meta.Filename,
			}).Post("http://" + ip + ":" + wxPort + "/SendFileMsg")
		}

		if err != nil {
			_ = tx.Rollback()
			continue
		} else {
			_, err := tx.MessageRecords.UpdateOneID(v.ID).SetStatus(3).SetSendTime(time.Now()).Save(ctx)
			if err != nil {
				_ = tx.Rollback()
			} else {
				_ = tx.Commit()
			}
		}

		time.Sleep(time.Duration(60/p.Number) * time.Second)
	}

	finishTime := time.Now()
	l.Logger.Infof("This process cost %v", finishTime.Sub(startTime).String())
	return
}