jx-callback/controllers/event_controller.go

package controllers

import (
	"fmt"
	"io"
	"log"
	"net/http"
	"os"
	"path"
	"time"

	"git.rosy.net.cn/jx-callback/business/jxstore/event"

	"git.rosy.net.cn/jx-callback/business/jxutils"
	"git.rosy.net.cn/jx-callback/business/jxutils/jxcontext"

	"git.rosy.net.cn/jx-callback/globals"

	"git.rosy.net.cn/jx-callback/business/model/dao"

	"git.rosy.net.cn/jx-callback/business/model"

	"github.com/gorilla/websocket"

	"git.rosy.net.cn/baseapi/utils"

	"github.com/astaxie/beego"
)

// 操作事件明细相关API
type EventController struct {
	beego.Controller
}

//连接的客户端,吧每个客户端都放进来
var clients = make(map[int]map[string]*websocket.Conn)
var clientsHeart = make(map[string]*websocket.Conn)

//广播频道(通道)
var broadcast = make(chan *model.ImMessageRecord)

// 配置升级程序(升级为websocket)
var upgrader = websocket.Upgrader{}

// 定义我们的消息对象
type Message struct {
	Data interface{} `json:"data"`
}

const (
	audioPath = "/jxdata/cthrgw/dist/audio/"
)

// @Title 测试websocket
// @Description 测试websocket
// @Success 200 {object} controllers.CallResult
// @Failure 200 {object} controllers.CallResult
// @router /TestWebsocket [get]
func (c *EventController) TestWebsocket() {
	// 解决跨域问题(微信小程序)
	upgrader.CheckOrigin = func(r *http.Request) bool {
		return true
	}
	//升级将HTTP服务器连接升级到WebSocket协议。
	//responseHeader包含在对客户端升级的响应中
	//请求。使用responseHeader指定Cookie（设置Cookie）和
	//应用程序协商的子目录（Sec WebSocket协议）。
	//如果升级失败，则升级将向客户端答复一个HTTP错误
	ws, err := upgrader.Upgrade(c.Ctx.ResponseWriter, c.Ctx.Request, nil)
	if err != nil {
		log.Fatal(err)
	}
	defer ws.Close()
	var (
		userID     = c.GetString("userID")
		clientUser = make(map[string]*websocket.Conn)
	)
	if userID == "" {
		return
	}
	//将当前客户端放入map中
	messageGroups, _ := dao.GetUserMessageGroups(dao.GetDB(), userID)
	if len(messageGroups) == 0 {
		return
	}
	clientUser[userID] = ws
	clientsHeart[userID] = ws
	for _, v := range messageGroups {
		if len(clients[v.GroupID]) > 0 {
			clients[v.GroupID][userID] = ws
		} else {
			clients[v.GroupID] = clientUser
		}
	}
	db := dao.GetDB()
	if globals.IsProductEnv() {
		_, _, err = jxcontext.New(nil, c.GetString("token"), c.Ctx.ResponseWriter, c.Ctx.Request)
		if err != nil {
			msg := &CallResult{
				Code: model.ErrCodeGeneralFailed,
				Desc: err.Error(),
			}
			ws.WriteJSON(&msg)
		}
		globals.SugarLogger.Debugf("TestWebsocket connection...")
	}
	globals.SugarLogger.Debugf("userID :%v ,clients :%v", userID, utils.Format4Output(clients, false))
	c.EnableRender = false //Beego不启用渲染

	var s *model.ImMessageRecord
	for {
		//接收客户端的消息
		err := ws.ReadJSON(&s)
		if err != nil {
			globals.SugarLogger.Debugf("页面可能断开啦 ws.ReadJSON error: %v", err.Error())
			for k, _ := range clients {
				delete(clients[k], userID)
			}
			delete(clientsHeart, userID)
			// delete(clients, ws) //删除map中的客户端
			break //结束循环
		} else {
			//接受消息 业务逻辑
			broadcast <- s
			if s.GroupID != 0 {
				//发聊天消息时，这个组所有的成员包括创建者都在userIDs里
				userIDs := []string{}
				if results, err := dao.GetMessageGroups(db, "", s.GroupID, 0, true, ""); err == nil {
					for _, v := range results {
						userIDs = append(userIDs, v.UserID)
						for _, vv := range v.MessageGroupMembers {
							userIDs = append(userIDs, vv.UserID)
						}
					}
				}
				//如果这些人不在这个组的ws池子里就打上未读标记
				for _, v := range userIDs {
					if clientsHeart[v] == nil {
						messageGroupReads, _ := dao.GetMessageGroupRead(db, v, s.GroupID)
						for _, vv := range messageGroupReads {
							vv.UnReadCount++
							dao.UpdateEntity(db, vv, "UnReadCount")
						}
					}
				}
			}
			utils.CallFuncAsync(func() {
				if s.GroupID != 0 {
					dao.WrapAddIDCULDEntity(s, "")
					dao.CreateEntity(db, s)
				}
			})
		}
	}
}

func init() {
	go handleMessages()
}

//广播推送消息
func handleMessages() {
	for {
		//读取通道中的消息
		msg := <-broadcast
		if msg.GroupID == 0 {
			// globals.SugarLogger.Debugf("heart %v", utils.Format4Output(msg, false))
			if clientsHeart[msg.UserID] != nil {
				if err := clientsHeart[msg.UserID].WriteJSON(&model.ImMessageRecord{
					Key: "pang",
				}); err != nil {
					globals.SugarLogger.Debugf("heart client.WriteJSON error: %v", err)
					clientsHeart[msg.UserID].Close() //关闭
					delete(clientsHeart, msg.UserID)
				}
			}
		} else {
			globals.SugarLogger.Debugf("clients len %v", len(clients))
			//循环map客户端
			for userID, client := range clients[msg.GroupID] {
				//把通道中的消息发送给客户端
				user, err := dao.GetUser(dao.GetDB(), msg.UserID)
				if err == nil {
					msg.UserInfo = user
				}
				globals.SugarLogger.Debugf("msg %v", utils.Format4Output(msg, false))
				err = client.WriteJSON(msg)
				if err != nil {
					globals.SugarLogger.Debugf("client.WriteJSON error: %v", err)
					client.Close() //关闭
					delete(clients[msg.GroupID], userID)
					// delete(clients, client) //删除map中的客户端
				}
			}
		}
	}
}

// @Title 查询聊天记录
// @Description 查询聊天记录
// @Param	token			header		string	true	"认证token"
// @Param	groupID			query		int		true	"组ID"
// @Param	fromTime		query		string	false	"开始时间"
// @Param	toTime			query		string	false	"结束时间"
// @Param	offset			query		int		false	"门店列表起始序号（以0开始，缺省为0）"
// @Param	pageSize		query		int		false	"门店列表页大小（缺省为50，-1表示全部）"
// @Success 200 {object} controllers.CallResult
// @Failure 200 {object} controllers.CallResult
// @router /GetImMessageRecord [get]
func (c *EventController) GetImMessageRecord() {
	c.callGetImMessageRecord(func(params *tEventGetImMessageRecordParams) (retVal interface{}, errCode string, err error) {
		var db = dao.GetDB()
		retVal, err = dao.GetImMessageRecord(db, params.GroupID, "", 0, -1, utils.Str2Time(params.FromTime), utils.Str2Time(params.ToTime), params.Offset, params.PageSize)
		//清除此用户组所有的未读标记
		if messageGroupReads, err := dao.GetMessageGroupRead(db, params.Ctx.GetUserID(), params.GroupID); err == nil {
			for _, v := range messageGroupReads {
				v.UnReadCount = 0
				dao.UpdateEntity(db, v, "UnReadCount")
			}
		}
		return retVal, "", err
	})
}

// @Title 用户未读消息设置
// @Description 用户未读消息设置，用户在退出登录，ws关闭，以及关闭小程序或app时调用
// @Param	token			header		string	true	"认证token"
// @Param	payload			query		string	true	"messageGroupRead 格式[{"groupID":,"unReadCount":}]"
// @Success 200 {object} controllers.CallResult
// @Failure 200 {object} controllers.CallResult
// @router /UpdateUserMessageGroupRead [post]
func (c *EventController) UpdateUserMessageGroupRead() {
	c.callUpdateUserMessageGroupRead(func(params *tEventUpdateUserMessageGroupReadParams) (retVal interface{}, errCode string, err error) {
		var reads []*model.MessageGroupRead
		if err = jxutils.Strings2Objs(params.Payload, &reads); err == nil {
			err = event.UpdateUserMessageGroupRead(params.Ctx, reads)
		}
		return retVal, "", err
	})
}

// @Title 发送聊天消息（限定系统消息）
// @Description 发送聊天消息（限定系统消息）
// @Param	token			header		string	true	"认证token"
// @Param	payload		 	formData	string	true	"immessageRecord 类型"
// @Success 200 {object} controllers.CallResult
// @Failure 200 {object} controllers.CallResult
// @router /SendSysMessage [post]
func (c *EventController) SendSysMessage() {
	c.callSendSysMessage(func(params *tEventSendSysMessageParams) (retVal interface{}, errCode string, err error) {
		var imMessageRecord *model.ImMessageRecord
		if err = jxutils.Strings2Objs(params.Payload, &imMessageRecord); err == nil {
			err = event.SendSysMessage(params.Ctx, imMessageRecord)
		}
		return retVal, "", err
	})
}

// @Title 创建聊天组
// @Description 创建聊天组
// @Param	token				header		string	true	"认证token"
// @Param	userID		 		formData	string	true	"创建者id"
// @Param	userID2		 		formData	string	false	"被拉的id 如果userID2为空就默认为是创建的群聊"
// @Param	name		 		formData	string	false	"如果是群聊，则要传入群名"
// @Param	dividePercentage	formData	int		false	"如果是群聊，则要传入分成比例"
// @Param	quitPrice			formData	int		false	"如果是群聊，则要传入退团金额"
// @Success 200 {object} controllers.CallResult
// @Failure 200 {object} controllers.CallResult
// @router /CreateMessageGroup [post]
func (c *EventController) CreateMessageGroup() {
	c.callCreateMessageGroup(func(params *tEventCreateMessageGroupParams) (retVal interface{}, errCode string, err error) {
		retVal, err = event.CreateMessageGroup(params.Ctx, params.UserID, params.UserID2, params.Name, params.DividePercentage, params.QuitPrice)
		return retVal, "", err
	})
}

// @Title 查询某个用户所有聊天组
// @Description 查询某个用户所有聊天组
// @Param	token			header		string	true	"认证token"
// @Param	userID		 	query		string	true	"userid"
// @Success 200 {object} controllers.CallResult
// @Failure 200 {object} controllers.CallResult
// @router /GetMessageGroupByUser [get]
func (c *EventController) GetMessageGroupByUser() {
	c.callGetMessageGroupByUser(func(params *tEventGetMessageGroupByUserParams) (retVal interface{}, errCode string, err error) {
		retVal, err = event.GetMessageGroupByUser(params.Ctx, params.UserID)
		return retVal, "", err
	})
}

// @Title 查询聊天组
// @Description 查询聊天组
// @Param	token			header		string	true	"认证token"
// @Param	groupID		 	query		int		true	"groupID"
// @Param	isMember		query		bool	true	"是否查询组员"
// @Success 200 {object} controllers.CallResult
// @Failure 200 {object} controllers.CallResult
// @router /GetMessageGroups [get]
func (c *EventController) GetMessageGroups() {
	c.callGetMessageGroups(func(params *tEventGetMessageGroupsParams) (retVal interface{}, errCode string, err error) {
		retVal, err = dao.GetMessageGroups(dao.GetDB(), "", params.GroupID, model.GroupTypeMulit, params.IsMember, "")
		return retVal, "", err
	})
}

// @Title 加入用户组
// @Description 加入用户组
// @Param	token			header		string	true	"认证token"
// @Param	groupID		 	formData	int		true	"组号"
// @Param	userID		 	formData	string	true	"被邀请人ID"
// @Success 200 {object} controllers.CallResult
// @Failure 200 {object} controllers.CallResult
// @router /AddMessageGroup [post]
func (c *EventController) AddMessageGroup() {
	c.callAddMessageGroup(func(params *tEventAddMessageGroupParams) (retVal interface{}, errCode string, err error) {
		err = event.AddMessageGroup(params.Ctx, params.GroupID, params.UserID)
		return retVal, "", err
	})
}

// @Title 修改群组信息
// @Description 修改群组信息
// @Param	token			header		string	true	"认证token"
// @Param	groupID		 	formData	int		true	"组号"
// @Param	payload		 	formData	string	true	"群组 payload"
// @Success 200 {object} controllers.CallResult
// @Failure 200 {object} controllers.CallResult
// @router /UpdateMessageGroup [put]
func (c *EventController) UpdateMessageGroup() {
	c.callUpdateMessageGroup(func(params *tEventUpdateMessageGroupParams) (retVal interface{}, errCode string, err error) {
		payload := make(map[string]interface{})
		if err = utils.UnmarshalUseNumber([]byte(params.Payload), &payload); err == nil {
			retVal, err = event.UpdateMessageGroup(params.Ctx, params.GroupID, payload)
		}
		return retVal, "", err
	})
}

// @Title 退出用户组（踢人）
// @Description 退出用户组（踢人）
// @Param	token			header		string	true	"认证token"
// @Param	groupID		 	formData	int		true	"组号"
// @Param	userID		 	formData	string	true	"userID"
// @Success 200 {object} controllers.CallResult
// @Failure 200 {object} controllers.CallResult
// @router /DeleteMessageGroup [post]
func (c *EventController) DeleteMessageGroup() {
	c.callDeleteMessageGroup(func(params *tEventDeleteMessageGroupParams) (retVal interface{}, errCode string, err error) {
		errCode, err = event.DeleteMessageGroup(params.Ctx, params.GroupID, params.UserID)
		return retVal, errCode, err
	})
}

// @Title 上传图片
// @Description 上传图片
// @Param	token			header		string	true	"认证token"
// @Success 200 {object} controllers.CallResult
// @Failure 200 {object} controllers.CallResult
// @router /UploadImg [post]
func (c *EventController) UploadImg() {
	c.callUploadImg(func(params *tEventUploadImgParams) (retVal interface{}, errCode string, err error) {
		file, head, err := c.GetFile("rsmImg")
		defer file.Close()
		if head.Size > 1024*1024*5 {
			err = fmt.Errorf("图片太大，请重新选择！")
		}
		if path.Ext(head.Filename) != ".jpg" && path.Ext(head.Filename) != ".png" {
			err = fmt.Errorf("不支持的图片格式，请重新选择！")
		}
		if err != nil {
			return retVal, "", err
		}
		fileName := utils.GetUUID() + "_" + time.Now().Format("20060102") + path.Ext(head.Filename)
		c.SaveToFile("rsmImg", "/jxdata/cthrgw/dist/img/"+fileName)
		return "https://www.jxcs.net/img/" + fileName, "", err
	})
}

// @Title 上传音频
// @Description 上传音频
// @Param	token			header		string	true	"认证token"
// @Success 200 {object} controllers.CallResult
// @Failure 200 {object} controllers.CallResult
// @router /UploadAudio [post]
func (c *EventController) UploadAudio() {
	c.callUploadAudio(func(params *tEventUploadAudioParams) (retVal interface{}, errCode string, err error) {
		var (
			timeStr  = time.Now().Format("20060102")
			filePath = audioPath + timeStr
		)
		files := c.Ctx.Request.MultipartForm.File["rsmAudio"]
		head := files[0]
		file, err := head.Open()
		defer file.Close()
		if path.Ext(head.Filename) != ".mp3" {
			err = fmt.Errorf("不支持的音频格式，请重新选择！")
		}
		if err != nil {
			return retVal, "", err
		}
		fileName := utils.GetUUID() + "_" + timeStr + path.Ext(head.Filename)
		if _, err = os.Stat(filePath); os.IsNotExist(err) {
			os.MkdirAll(filePath, os.ModePerm)
			os.Chmod(filePath, os.ModePerm)
		}
		f, err := os.OpenFile(filePath+"/"+fileName, os.O_WRONLY|os.O_CREATE|os.O_TRUNC, 0666)
		if err != nil {
			return retVal, "", err
		}
		defer f.Close()
		io.Copy(f, file)

		return &model.ImMessageRecord{
			Content: "https://www.jxcs.net/audio/" + timeStr + "/" + fileName,
		}, "", err
	})
}