fonchain-fiee/pkg/service/asChat/handler.go

// package asChat -----------------------------
// @file      : handler.go
// @author    : JJXu
// @contact   : wavingbear@163.com
// @time      : 2022/10/23 11:13:43
// -------------------------------------------
package asChat

import (
	"bytes"
	"context"
	"crypto/md5"
	"encoding/hex"
	"errors"
	"fmt"
	"fonchain-fiee/api/account"
	"fonchain-fiee/api/accountFiee"
	"fonchain-fiee/pkg/common/ws"
	"fonchain-fiee/pkg/e"
	"fonchain-fiee/pkg/service"
	"fonchain-fiee/pkg/service/upload"
	"fonchain-fiee/pkg/utils"
	"github.com/fonchain/utils/voice"
	"github.com/gin-gonic/gin"
	"github.com/gorilla/websocket"
	uuid "github.com/satori/go.uuid"
	"go.uber.org/zap"
	"io"
	"log"
	"path"
	"slices"
	"sort"
	"strconv"
	"strings"
	"time"
)

var ChatHandlerIns = ChatHandler{
	cache: ChatCache{newMessageStatExpireAfter: 10 * time.Minute},
}

type ChatHandler struct {
	cache ChatCache
}

func (cr ChatHandler) Connection(c *gin.Context) {
	conn, err := ws.UpGrader.Upgrade(c.Writer, c.Request, nil)
	conn.SetReadDeadline(time.Now().Add(time.Second * 10))
	if err != nil {
		log.Fatal("无法升级为websocket连接", zap.Error(err))
		c.String(500, "无法转为websocket连接")
		return
	}
	defer func() {
		if conn != nil {
			conn.Close()
		}
	}()
	_, byteData, err := conn.ReadMessage()
	if err != nil {
		_ = conn.WriteMessage(websocket.TextMessage, ws.WsErrorConnection("null", err.Error(), "conn.ReadMessag1"))
		return
	}
	fmt.Println("22222222222222,AuthorizationVerify")
	var ok bool
	var userInfo *accountFiee.ChatUserData
	userInfo, ok, err = ws.AuthorizationVerify(byteData)
	if err != nil {
		_ = conn.WriteMessage(websocket.TextMessage, ws.WsErrorConnection("null", err.Error(), "AuthorizationVerify2"))
		return
	}
	if !ok {
		_ = conn.WriteMessage(websocket.TextMessage, ws.WsErrorConnection("null", "登录状态失效", "AuthorizationVerify2.1"))
		return
	}
	fmt.Println("33333333333333,RecountNewMessageTotal")
	conn.SetReadDeadline(time.Time{})
	go cr.cache.RecountNewMessageTotal(userInfo.ID)

	fmt.Println("44444444444444,ws.NewClient")
	//注册ws客户端，并发送clientId给ws客户端
	var cli = ws.NewClient(userInfo.ID, "", conn, ChatRoom)
	cli.Waiter = userInfo.Role == 2
	fmt.Println("55555555555555,GetUserSession")
	//查询是否有历史的sessionId
	cli.SessionId = cr.cache.GetUserSession(userInfo.ID)
	ChatRoom.Register(cli)
	cr.cache.SaveUserSession(userInfo.ID, cli.SessionId)
	fmt.Println("66666666666666666666666666")
	go cli.WriteWait()
	cli.Send <- WsMessageRegisterCallback(cli.ClientId, cli.SessionId)
	fmt.Println("777777777777777777777777")
	// 处理websocket连接的逻辑
	ctx, _ := context.WithCancel(context.Background())
	cli.Reading(ctx, HandleMessage)
	fmt.Println("88888888888888888888888888")
	select {
	case <-ctx.Done():
		return
	}
}

func (cr ChatHandler) NewMessage(c *gin.Context) {
	var request NewMessageRequest
	if err := c.ShouldBindJSON(&request); err != nil {
		service.Error(c, err)
		return
	}
	if request.SessionId == "" {
		service.Error(c, errors.New("sessionId不能为空"))
		return
	}
	if request.MsgType == 0 {
		service.Error(c, errors.New("msgType不能为空"))
		return
	}
	fmt.Println("NewMessage 1111111111111111111111111111111")
	//获取用户信息
	tokenResult := asAccount.GetUserInfoWithTokenV2(c)
	if tokenResult.Err != nil {
		service.Error(c, errors.New("未登录"))
		return
	}
	fmt.Println("NewMessage 22222222222222222222222222222222222")
	//存储入库
	var userName = "未知"
	if request.Waiter {
		accountDetail, err := service.AccountProvider.Info(c, &account.InfoRequest{ID: uint64(tokenResult.UserInfo.MgmtAccId), Domain: config.Domain})
		if err != nil {
			service.Error(c, errors.New("用户信息获取失败"))
			return
		}
		userName = accountDetail.Info.NickName
	} else {
		if tokenResult.UserInfo.RealNameInfo != nil {
			userName = tokenResult.UserInfo.RealNameInfo.Name
		}
	}
	fmt.Println("NewMessage 3333333333333333333333333333333333")
	var data = accountFiee.ChatRecordData{
		SessionId:  request.SessionId,
		UserId:     tokenResult.UserInfo.ID,
		Name:       userName,
		Avatar:     "",
		MsgType:    request.MsgType,
		Content:    request.Message.Text,
		LocalStamp: request.LocalStamp,
		Medias:     nil,
	}
	if len(request.Message.Media) > 0 {
		for _, media := range request.Message.Media {
			data.Medias = append(data.Medias, &accountFiee.ChatMediaData{
				ID: media.MediaId,
			})
		}
	}
	fmt.Println("NewMessage 4444444444444444444444444444444444")
	resp, err := service.AccountFieeProvider.CreateChatRecord(c, &data)
	if err != nil {
		service.Error(c, errors.New("创建失败"))
		return
	}
	fmt.Printf("CreateChatRecord resp:%+v\n", resp)
	//录入缓存
	err = cr.cache.AddChatRecord(request.SessionId, resp.Data)
	if err != nil {
		service.Error(c, errors.New("创建失败"))
		return
	}
	fmt.Println("NewMessage 5 消息数量+1")
	//新消息数量统计+1
	noticeUserId := ChatRoom.GetUserIdInSession(request.SessionId, tokenResult.UserInfo.ID)
	fmt.Println("NewMessage 5.1 消息数量配置结束")
	fmt.Printf("noticeUserId %+v\n", noticeUserId)
	for _, userId := range noticeUserId {
		fmt.Println("userId")
		cr.cache.IncreaseNewMessageTotal(userId, request.SessionId)
	}
	fmt.Println("NewMessage 6")
	//发送websocket消息提醒通知
	var notice = MessageListType{}
	notice.BuildMessage(resp.Data)
	fmt.Printf("ws消息提醒:%+v\n", notice)
	_, err = ChatRoom.SendSessionMessage(tokenResult.UserInfo.ID, request.SessionId, ws.NewChatMsgType, notice)
	if err != nil {
		log.Fatal("发送新消息通知失败", zap.Error(err), zap.Any("notice", notice))
	}
	fmt.Println("NewMessage 7 -end")
	//发送app推送(无横幅推送)
	go func() {
		omitMessage := ""
		switch request.MsgType {
		case accountFiee.MsgType_TextMsgType:
			runMsg := []rune(request.Text)
			if len(runMsg) > 15 {
				omitMessage = string(runMsg[:15]) + "..."
			} else {
				omitMessage = request.Text
			}
		case accountFiee.MsgType_ImageMsgType:
			omitMessage = "[图片]"
		case accountFiee.MsgType_AudioMsgType:
			omitMessage = "[音频]"
		case accountFiee.MsgType_VideoMsgType:
			omitMessage = "[视频]"
		default:
			omitMessage = "新消息请查收"
		}
		for _, userId := range noticeUserId {
			_ = asPusher.NewArtistinfoUniPush().NewChatMessageNotice(userId, omitMessage)
		}
	}()
	service.Success(c)
}

func (cr ChatHandler) MessageList(c *gin.Context) {
	var request MessageListRequest
	if err := c.ShouldBindJSON(&request); err != nil {
		service.Error(c, err)
		return
	}
	domain := c.GetHeader("domain")
	if (request.Direction == 0 && request.Recent == false) || (request.Direction > 0 && request.Recent == true) {
		service.Error(c, errors.New("组合条件校验失败"))
		return
	}
	if request.SessionId == "" {
		service.Error(c, errors.New("sessionId不能为空"))
		return
	}
	if request.PageSize < -1 {
		service.Error(c, errors.New("pageSize校验错误"))
		return
	}
	var resp = make([]*MessageListType, 0)
	if request.CurrentId == 0 && request.Direction == 1 {
		service.Success(c, resp)
		return
	}
	tokenResult := asAccount.GetUserInfoWithTokenV2(c)
	if tokenResult.Err != nil {
		service.ErrorWeb(c, e.NotLogin, tokenResult.Err.Error())
		return
	}
	//if request.SessionId == "" {
	//	request.SessionId = cr.cache.GetUserSession(tokenResult.UserInfo.ID)
	//	if request.SessionId == "" {
	//		service.Success(c, resp)
	//		return
	//	}
	//}
	//messages := cr.cache.GetChatRecord(request.SessionId)
	messages := []*accountFiee.ChatRecordData{}
	var returnDataIdList = make([]int64, 0)
	defer func() {
		//获取最新数据时，重置新消息数量统计
		if request.Direction == 2 || request.Recent {
			cr.cache.ResetNewMessageTotal(tokenResult.UserInfo.ID, request.SessionId)
		}
		//设置消息已被客服阅读,当客服重新通过通过websocket连接时，这些消息将不被纳入新消息数量统计
		if len(returnDataIdList) > 0 && domain == "fontree" {
			for _, hasReadId := range returnDataIdList {
				for i, message := range messages {
					if message.ID == hasReadId {
						messages[i].WaiterRead = 1
					}
				}
			}
			err := cr.cache.CoverChatRecord(request.SessionId, messages)
			if err != nil {
				log.Fatal("设置消息已读失败", zap.Error(err))
			}
			for _, v := range messages {
				_, err = service.AccountFieeProvider.SaveChatRecord(context.Background(), v)
				if err != nil {
					log.Fatal("设置消息已读失败", zap.Error(err))
				}
			}
		}
	}()
	if len(messages) == 0 {
		//从数据库获取
		recordResp, err := service.AccountFieeProvider.GetChatRecordList(c, &accountFiee.GetChatRecordListRequest{
			Query:    &accountFiee.ChatRecordData{SessionId: request.SessionId},
			Page:     1,
			PageSize: -1,
			//Where:    fmt.Sprintf("id %s %d", utils.IfGec(request.Direction == 1, "<", ">"), request.CurrentId),
		})
		if err != nil {
			service.Error(c, err)
			return
		}
		messages = recordResp.List
		err = cr.cache.CoverChatRecord(request.SessionId, messages)
		if err != nil {
			log.Fatal("覆盖聊天记录失败", zap.Error(err))
		}
	}
	if request.Recent {
		if int64(len(messages)) >= request.PageSize {
			messages = messages[len(messages)-int(request.PageSize):]
		}
		var now = time.Now()
		for _, message := range messages {
			if request.InHour > 0 {
				messageCreatedAt, _ := stime.StringToTime(message.CreatedAt)
				if now.Sub(*messageCreatedAt) >= request.InHour*time.Hour {
					continue
				}
			}
			returnDataIdList = append(returnDataIdList, message.ID)
			var msg = &MessageListType{}
			msg.BuildMessage(message)
			resp = append(resp, msg)
		}
	} else {
		sort.Slice(messages, func(i, j int) bool {
			if request.Direction == 1 {
				return messages[i].ID < messages[j].ID
			} else {
				return messages[i].ID > messages[j].ID
			}
		})
		fmt.Printf("data is %+v\n", messages)
		total := 0
		for i, message := range messages {
			switch request.Direction {
			case 1: //向下查找，找比CurrentId大的数据
				if message.ID <= request.CurrentId {
					continue
				}
			case 2: //向上查找，找比CurrentId小的数据
				if message.ID >= request.CurrentId {
					continue
				}
			}
			message := message
			fmt.Println(i, message.ID)
			if request.PageSize != -1 && int64(total+1) > request.PageSize {
				continue
			}
			total++
			returnDataIdList = append(returnDataIdList, message.ID)
			var msg = &MessageListType{}
			msg.BuildMessage(message)
			resp = append(resp, msg)
		}
	}
	//二次排序
	sort.Slice(resp, func(i, j int) bool {
		return resp[i].ID < resp[j].ID
	})
	//优化空列表
	for i, v := range resp {
		if v.Message.Media == nil {
			resp[i].Message.Media = []MessageMedia{}
		}
	}
	service.Success(c, resp)
}

func (cr ChatHandler) Upload(c *gin.Context) {
	fmt.Println("111111111111")
	//获取用户信息
	accInfo := asAccount.GetUserInfoWithTokenV2(c)
	if accInfo.Err != nil {
		service.Error(c, accInfo.Err.Error())
		return
	}
	//获取文件对象
	file, err := c.FormFile("file")
	if err != nil {
		log.Fatal("ERROR: upload file failed. ", zap.Error(err))
		return
	}
	duration := c.PostForm("duration")
	fmt.Println(duration)
	ext := c.PostForm("ext")
	fileExt := strings.ToLower(path.Ext(file.Filename))
	if ext != "" {
		fileExt = ext
	}
	fileType := e.DetectFileTypeByExtension(fileExt)
	if fileType == e.Audio {
		if !slices.Contains([]string{".mp4", ".aac", ".mp3", ".opus", ".wav"}, fileExt) {
			service.Error(c, errors.New("不支持的格式"))
			return
		}
	}
	//计算md5
	tmp, err := file.Open()
	if err != nil {
		service.Error(c, errors.New("上传失败"))
		return
	}
	fileContent, err := io.ReadAll(tmp)
	if err != nil {
		service.Error(c, errors.New("文件读取失败"))
		return
	}
	hash := md5.New()
	_, err = hash.Write(fileContent)
	if err != nil {
		service.Error(c, errors.New("文件读取失败"))
		return
	}
	md5Bytes := hash.Sum(nil)                 // 获取 MD5 字节切片
	md5String := hex.EncodeToString(md5Bytes) // 转换为十六进制字符串表示
	//检查文件是否存在
	checkResp, err := service.AccountFieeProvider.GetChatMediaList(c, &accountFiee.GetChatMediaListRequest{Query: &accountFiee.ChatMediaData{Md5: md5String}, Page: 1, PageSize: 1})
	if err != nil {
		log.Fatal("md5查询附件失败", zap.Error(err))
	}
	if checkResp.Total > 0 {
		service.Success(c, checkResp.List[0])
		return
	}

	//文件不存在则上传文件
	filename, _ := uuid.NewV4()
	defer tmp.Close()
	fileBuffer := bytes.NewBuffer(fileContent)
	var bosUrl string
	bosUrl, err = upload.UploadWithBuffer(fileBuffer, fmt.Sprintf("%d/%v%v", accInfo.UserInfo.MgmtAccId, filename, fileExt))
	if err != nil {
		service.Error(c, err)
		return
	}
	//存到数据库
	var durationInt64, _ = strconv.ParseInt(duration, 10, 64)
	var mediaData = accountFiee.ChatMediaData{
		Url:      bosUrl,
		Md5:      md5String,
		Size:     fmt.Sprintf("%d", file.Size),
		Ext:      fileExt,
		Duration: durationInt64,
	}
	resp, err := service.AccountFieeProvider.CreateChatMedia(c, &mediaData)
	if err != nil {
		service.Error(c, err)
		return
	}
	service.Success(c, resp.Data)
}

func (cr ChatHandler) UserMessageStat(c *gin.Context) {
	//获取用户信息
	tokenResult := asAccount.GetUserInfoWithTokenV2(c)
	if tokenResult.Err != nil {
		service.Error(c, errors.New("未登录"))
		return
	}
	result := cr.cache.GetNewMessageStat(c, tokenResult.UserInfo.ID)
	if len(result) == 0 {
		service.Success(c, result)
		return
	}
	fmt.Printf("cache stat:%+v\n", result)
	//获取实名信息
	var protoReq = accountFiee.GetChatUserListRequest{
		Page:     1,
		PageSize: int64(len(result)),
	}
	for i, item := range result {
		if item.UserId == 0 {
			sessionId, _ := strconv.Atoi(item.SessionId)
			item.UserId = int64(sessionId)
			result[i].UserId = int64(sessionId)
		}
		protoReq.UserIdIn = append(protoReq.UserIdIn, item.UserId)
	}
	fmt.Printf("protoReq.UserIdIn:%+v\n", protoReq.UserIdIn)
	listRes, err := service.AccountFieeProvider.GetChatUserList(c, &protoReq)
	if err != nil {
		service.Error(c, err)
		return
	}
	fmt.Printf("GetChatUserList:%+v\n", listRes)
	for i, item := range result {
		for _, user := range listRes.List {
			if item.UserId == user.UserId {
				user := user
				result[i].Name = user.Name
				result[i].ArtistUid = user.ArtistUid
				break
			}
		}
		if result[i].Name == "" {
			result[i].Name = beautifulZeroName(result[i].Name, result[i].UserId)
		}
	}
	reverse(result)
	service.Success(c, result)
}
func reverse(slice []UserMsgStatic) {
	for i, j := 0, len(slice)-1; i < j; i, j = i+1, j-1 {
		slice[i], slice[j] = slice[j], slice[i]
	}
}
func (cr ChatHandler) VoiceToText(c *gin.Context) {
	var req VoiceToTextRequest
	if err := c.ShouldBindJSON(&req); err != nil {
		service.Error(c, err)
		return
	}
	detail, err := service.AccountFieeProvider.GetChatMediaDetail(c, &accountFiee.GetChatMediaByIdRequest{Id: req.MediaId})
	if err != nil {
		service.Error(c, err)
		return
	}
	if detail.ConvText != "" {
		service.Success(c, map[string]string{"convText": detail.ConvText})
		return
	}
	voiceApi := voice.NewVoiceApi()
	detail.ConvText, err = voiceApi.ToTextFromUrl(detail.Url)
	if err != nil {
		service.Error(c, errors.New("语音转文字失败"))
		return
	}
	defer func() {
		service.AccountFieeProvider.UpdateChatMedia(context.Background(), detail)
	}()
	service.Success(c, map[string]string{"convText": detail.ConvText})
}

func (cr ChatHandler) ArtistDetail(c *gin.Context) {
	var req ArtistInfoRequest
	if err := c.ShouldBindJSON(&req); err != nil {
		service.Error(c, err)
		return
	}
	if req.UserId == 0 {
		service.Success(c, ArtistInfo{})
		return
	}
	detail, err := service.GrpcArtistInfoUserImpl.FindUsersUserView(c, &artistInfoUser.FindUsersRequest{UserId: req.UserId})
	if err != nil {
		service.Error(c, err)
		return
	}
	var (
		tnum        string
		artistName  string
		age         int64
		sex         string
		nativePlace string
		telNum      string
		recentPhoto string
	)
	if len(detail.Data) > 0 {
		tnum = detail.Data[0].Tnum
		artistName = beautifulZeroName(detail.Data[0].RealName, req.UserId)
		age = detail.Data[0].Age
		sex = detail.Data[0].Sex
		nativePlace = detail.Data[0].NativePlace
		telNum = detail.Data[0].TelNum
		recentPhoto = detail.Data[0].Photo
	}
	resp := ArtistInfo{
		Tnum:        tnum,
		ArtistName:  artistName,
		Age:         age,
		Sex:         sex,
		NativePlace: nativePlace,
		TelNum:      telNum,
		RecentPhoto: recentPhoto,
	}
	service.Success(c, resp)
}

// 对没有名字的name进行优化
func beautifulZeroName(name string, userId int64) string {
	return utils.IfGec(name == "", fmt.Sprintf("未实名用户:%d", userId), name)
}