wechat-api/internal/logic/batch_msg/create_batch_msg_logic.go

package batch_msg

import (
	"context"
	"encoding/json"
	"errors"
	"strings"
	"time"
	"wechat-api/ent/custom_types"

	"wechat-api/ent"
	"wechat-api/ent/contact"
	"wechat-api/ent/label"
	"wechat-api/ent/labelrelationship"
	"wechat-api/internal/svc"
	"wechat-api/internal/types"
	"wechat-api/internal/utils/dberrorhandler"

	"github.com/suyuan32/simple-admin-common/msg/errormsg"
	"github.com/suyuan32/simple-admin-common/utils/uuidx"
	"github.com/zeromicro/go-zero/core/logx"
)

type CreateBatchMsgLogic struct {
	ctx    context.Context
	svcCtx *svc.ServiceContext
	logx.Logger
}

func NewCreateBatchMsgLogic(ctx context.Context, svcCtx *svc.ServiceContext) *CreateBatchMsgLogic {
	return &CreateBatchMsgLogic{
		ctx:    ctx,
		svcCtx: svcCtx,
		Logger: logx.WithContext(ctx),
	}
}

func (l *CreateBatchMsgLogic) CreateBatchMsg(req *types.BatchMsgInfo) (*types.BaseMsgResp, error) {
	var err error
	organizationId := l.ctx.Value("organizationId").(uint64)

	all := false
	for _, labelName := range req.Labels {
		if strings.EqualFold(labelName, "all") || strings.EqualFold(labelName, "全部") {
			all = true
			tagstring := "all"
			req.Tag = &tagstring
		}
	}

	l.Logger.Infof("CreateBatchMsgLogic: %v", *req)

	startTime := time.Now()
	// req.StartTimeStr 不为nil并且不为空
	if req.StartTimeStr != nil && *req.StartTimeStr != "" {
		// 将 req.StartTimeStr 转换为 req.StartTime
		startTime, err = time.Parse("2006-01-02 15:04:05", *req.StartTimeStr)
		if err != nil {
			// 处理错误，例如打印错误并返回
			l.Logger.Errorf("时间字符串转换错误: %v", err)
			return nil, err
		}
	}

	// 定时发送时间
	var sendTime *time.Time
	if req.SendTimeStr != nil && *req.SendTimeStr != "" {
		sendTimeParse, err := time.Parse("2006-01-02 15:04:05", *req.SendTimeStr)
		if err != nil {
			// 处理错误，例如打印错误并返回
			l.Logger.Errorf("时间字符串转换错误: %v", err)
			return nil, err
		}
		sendTime = &sendTimeParse
	}

	// 把 req.Msg 字符串的内容 json_decode 到 msgArray
	// 然后再把每一条信息都给所有指定的用户记录到 message_records 表
	var msgArray []custom_types.Action

	if req.Msg != nil && *req.Msg != "" {
		err = json.Unmarshal([]byte(*req.Msg), &msgArray)
		if err != nil {
			return nil, errors.New("解析JSON失败")
		}
	}

	var msgActionList []custom_types.Action
	if len(msgArray) > 0 {
		msgActionList = make([]custom_types.Action, len(msgArray))
		for i, msg := range msgArray {
			if msg.Type == 1 {
				msgActionList[i] = custom_types.Action{
					Type:    msg.Type,
					Content: msg.Content,
				}
			} else {
				msgActionList[i] = custom_types.Action{
					Type:    msg.Type,
					Content: msg.Content,
					Meta: &custom_types.Meta{
						Filename: msg.Meta.Filename,
					},
				}
			}
		}
	}

	tagstring := strings.Join(req.Labels, ",")
	req.Tag = &tagstring

	userlist := make([]*ent.Contact, 0)

	if all {
		// 获取 contact 表中  wx_wxid 等于 req.Fromwxid 的 type 为1或2的数据
		userlist, err = l.svcCtx.DB.Contact.Query().Where(contact.WxWxid(*req.Fromwxid), contact.TypeIn(1, 2)).All(l.ctx)
		if err != nil {
			return nil, dberrorhandler.DefaultEntError(l.Logger, err, req)
		}
	} else {
		// 获取 label 表中 name 为 tags的记录
		labids, err := l.svcCtx.DB.Label.Query().Where(label.NameIn(req.Labels...)).IDs(l.ctx)
		if err != nil {
			return nil, dberrorhandler.DefaultEntError(l.Logger, err, req)
		}
		//l.Logger.Infof("CreateBatchMsgLogic labids= %v", labids)

		// 获取 label_relationship 表中，label_id 等于 labids 的 contact_id
		labelrelationships, err := l.svcCtx.DB.LabelRelationship.Query().Where(labelrelationship.LabelIDIn(labids...)).All(l.ctx)
		if err != nil {
			return nil, dberrorhandler.DefaultEntError(l.Logger, err, req)
		}
		contact_ids := make([]uint64, 0)
		for _, labelrelationship := range labelrelationships {
			contact_ids = append(contact_ids, labelrelationship.ContactID)
		}
		//l.Logger.Infof("CreateBatchMsgLogic contact_ids= %v", contact_ids)

		if len(contact_ids) > 0 {
			// 获取 contact 表中 wx_wxid 等于 req.Fromwxid 并且 id 等于 contact_ids 并且 type 为1或2 的数据
			userlist, err = l.svcCtx.DB.Contact.Query().Where(contact.WxWxid(*req.Fromwxid), contact.IDIn(contact_ids...), contact.TypeIn(1, 2)).All(l.ctx)
			if err != nil {
				return nil, dberrorhandler.DefaultEntError(l.Logger, err, req)
			}
			//l.Logger.Infof("CreateBatchMsgLogic userlist= %v", userlist)
		}
	}
	total := int32(len(userlist))

	if total == 0 {
		return &types.BaseMsgResp{Msg: errormsg.TargetNotFound}, nil
	}

	uuid := uuidx.NewUUID()
	batchNo := uuid.String()

	// 开始事务
	tx, err := l.svcCtx.DB.Tx(context.Background())
	if err != nil {
		return nil, dberrorhandler.DefaultEntError(l.Logger, err, req)
	}

	batchMsg, err := l.svcCtx.DB.BatchMsg.Create().
		SetNotNilBatchNo(&batchNo).
		SetNotNilFromwxid(req.Fromwxid).
		SetNotNilMsg(req.Msg).
		SetNotNilTag(req.Tag).
		SetTotal(total).
		SetNotNilTaskName(req.TaskName).
		SetNotNilStartTime(&startTime).
		SetNotNilSendTime(sendTime).
		SetNotNilType(req.Type).
		Save(l.ctx)

	if err != nil {
		_ = tx.Rollback()
		return nil, dberrorhandler.DefaultEntError(l.Logger, err, req)
	}

	// 批量记录信息到 message_records 表里
	for _, user := range userlist {
		// 每个用户的所有 信息 组合到一个数组里然后批量插入，减少DB的压力
		bulkCreate := make([]*ent.MessageRecordsCreate, 0)
		for idx, msg := range msgActionList {
			bulkCreate = append(bulkCreate, l.svcCtx.DB.MessageRecords.Create().
				SetStatus(1).
				SetNotNilBotWxid(req.Fromwxid).
				SetNotNilContactID(&user.ID).
				SetNotNilContactType(&user.Type).
				SetNotNilContentType(&msg.Type).
				SetNotNilContent(&msg.Content).
				SetNotNilMeta(msg.Meta).
				SetNotNilSendTime(sendTime).
				SetNotNilContactWxid(&user.Wxid). //接收方wxid
				SetSourceType(2).                 // 2:批量消息
				SetSourceID(batchMsg.ID).         // 批量消息ID，batch_msg 表主键
				SetSubSourceID(uint64(idx)).      // 用索引作为 sub_source_id
				SetOrganizationID(organizationId),
			)
		}
		err = tx.MessageRecords.CreateBulk(bulkCreate...).Exec(l.ctx)
		if err != nil {
			_ = tx.Rollback()
			return nil, dberrorhandler.DefaultEntError(l.Logger, err, req)
		}
	}
	_ = tx.Commit()

	return &types.BaseMsgResp{Msg: errormsg.CreateSuccess}, nil
}