Merge branch 'develop' of https://gitee.com/fusenpack/fusenapi into feature/auth

server/websocket/internal/logic/datatransferlogic.go

@@ -5,8 +5,6 @@ import (
 	"encoding/json"
 	"fmt"
 	"fusenapi/constants"
-	"fusenapi/initalize"
-	"fusenapi/model/gmodel"
 	"fusenapi/utils/auth"
 	"fusenapi/utils/id_generator"
 	"fusenapi/utils/websocket_data"
@@ -70,22 +68,20 @@ var (
 
 // 每个连接的连接基本属性
 type wsConnectItem struct {
-	conn           *websocket.Conn //websocket的连接
-	ctx            context.Context
-	rabbitMq       *initalize.RabbitMqHandle
-	allModels      *gmodel.AllModelsGen
-	closeChan      chan struct{}  //ws连接关闭chan
-	isClose        bool           //是否已经关闭
-	uniqueId       string         //ws连接唯一标识
-	inChan         chan []byte    //接受消息缓冲通道
-	outChan        chan []byte    //发送回客户端的消息
-	mutex          sync.Mutex     //互斥锁
-	userId         int64          //用户id
-	guestId        int64          //游客id
-	renderProperty renderProperty //扩展云渲染属性
+	conn           *websocket.Conn    //websocket的连接
+	logic          *DataTransferLogic //logic
+	closeChan      chan struct{}      //ws连接关闭chan
+	isClose        bool               //是否已经关闭
+	uniqueId       string             //ws连接唯一标识
+	inChan         chan []byte        //接受消息缓冲通道
+	outChan        chan []byte        //发送回客户端的消息
+	mutex          sync.Mutex         //互斥锁
+	userId         int64              //用户id
+	guestId        int64              //游客id
+	renderProperty renderProperty     //扩展云渲染属性
 }
 
-func (l *DataTransferLogic) DataTransfer(svcCtx *svc.ServiceContext, w http.ResponseWriter, r *http.Request) {
+func (l *DataTransferLogic) DataTransfer(w http.ResponseWriter, r *http.Request) {
 	//升级websocket
 	conn, err := upgrade.Upgrade(w, r, nil)
 	if err != nil {
@@ -98,7 +94,7 @@ func (l *DataTransferLogic) DataTransfer(svcCtx *svc.ServiceContext, w http.Resp
 		userInfo *auth.UserInfo
 		isAuth   bool
 	)
-	isAuth, userInfo = l.checkAuth(svcCtx, r)
+	isAuth, userInfo = l.checkAuth(r)
 	if !isAuth {
 		time.Sleep(time.Second * 1) //兼容下火狐
 		rsp := websocket_data.DataTransferData{
@@ -126,6 +122,8 @@ func (l *DataTransferLogic) DataTransfer(svcCtx *svc.ServiceContext, w http.Resp
 	go ws.sendLoop()
 	//操作连接中渲染任务的增加/删除
 	go ws.operationRenderTask()
+	//消费渲染缓冲队列
+	go ws.renderImage()
 	//心跳
 	ws.heartbeat()
 }
@@ -138,18 +136,17 @@ func (l *DataTransferLogic) setConnPool(conn *websocket.Conn, userInfo auth.User
 	uniqueId := l.getUniqueId(userInfo)
 	ws := wsConnectItem{
 		conn:      conn,
-		ctx:       l.ctx,
-		rabbitMq:  l.svcCtx.RabbitMq,
-		allModels: l.svcCtx.AllModels,
+		logic:     l,
 		uniqueId:  uniqueId,
 		closeChan: make(chan struct{}, 1),
-		inChan:    make(chan []byte, 1000),
-		outChan:   make(chan []byte, 1000),
+		inChan:    make(chan []byte, 100),
+		outChan:   make(chan []byte, 100),
 		userId:    userInfo.UserId,
 		guestId:   userInfo.GuestId,
 		renderProperty: renderProperty{
 			renderImageTask:        make(map[string]string),
 			renderImageTaskCtlChan: make(chan renderImageControlChanItem, 100),
+			renderChan:             make(chan []byte, 100),
 		},
 	}
 	//保存连接
@@ -178,9 +175,9 @@ func getUserPart(userId, guestId int64) string {
 }
 
 // 鉴权
-func (l *DataTransferLogic) checkAuth(svcCtx *svc.ServiceContext, r *http.Request) (isAuth bool, userInfo *auth.UserInfo) {
+func (l *DataTransferLogic) checkAuth(r *http.Request) (isAuth bool, userInfo *auth.UserInfo) {
 	// 解析JWT token,并对空用户进行判断
-	claims, err := svcCtx.ParseJwtToken(r)
+	claims, err := l.svcCtx.ParseJwtToken(r)
 	// 如果解析JWT token出错,则返回未授权的JSON响应并记录错误消息
 	if err != nil {
 		return false, nil
@@ -321,7 +318,7 @@ func (w *wsConnectItem) dealwithReciveData(data []byte) {
 	switch parseInfo.T {
 	//图片渲染
 	case constants.WEBSOCKET_RENDER_IMAGE:
-		w.renderImage(d)
+		w.sendToRenderChan(d)
 	//刷新重连请求恢复上次连接的标识
 	case constants.WEBSOCKET_REQUEST_REUSE_LAST_CONNECT:
 		w.reuseLastConnect(d)

server/websocket/internal/logic/mq_consumer.go

@@ -1,59 +0,0 @@
-package logic
-
-import (
-	"context"
-	"encoding/json"
-	"fusenapi/constants"
-	"fusenapi/utils/websocket_data"
-
-	"github.com/zeromicro/go-zero/core/logx"
-)
-
-// 消费渲染结果数据
-type MqConsumerRenderResult struct {
-}
-
-func (m *MqConsumerRenderResult) Run(ctx context.Context, data []byte) error {
-	defer func() {
-		if err := recover(); err != nil {
-			logx.Error("MqConsumerRenderResult panic:", err)
-		}
-	}()
-	logx.Info("接收到MqConsumerRenderResult数据:", string(data))
-	var parseInfo websocket_data.RenderImageNotify
-	if err := json.Unmarshal(data, &parseInfo); err != nil {
-		logx.Error("MqConsumerRenderResult data format err:", err)
-		return nil //不返回错误则就删掉该消息
-	}
-	//遍历websocket链接把数据传进去
-	mapConnPool.Range(func(key, value any) bool {
-		//断言连接
-		ws, ok := value.(wsConnectItem)
-		if !ok {
-			return true
-		}
-		//关闭标识
-		if ws.isClose {
-			return true
-		}
-		//查询有无该渲染任务
-		renderId, ok := ws.renderProperty.renderImageTask[parseInfo.TaskId]
-		if !ok {
-			return true
-		}
-		b := ws.respondDataFormat(constants.WEBSOCKET_RENDER_IMAGE, websocket_data.RenderImageRspMsg{
-			RenderId: renderId,
-			Image:    parseInfo.Image,
-		})
-		//删除对应的需要渲染的图片map
-		ws.renderProperty.renderImageTaskCtlChan <- renderImageControlChanItem{
-			Option:   0, //0删除 1添加
-			TaskId:   parseInfo.TaskId,
-			RenderId: renderId,
-		}
-		//发送数据到out chan
-		ws.sendToOutChan(b)
-		return true
-	})
-	return nil
-}

server/websocket/internal/logic/rendernotifylogic.go

@@ -0,0 +1,111 @@
+package logic
+
+import (
+	"fusenapi/constants"
+	"fusenapi/utils/auth"
+	"fusenapi/utils/basic"
+	"fusenapi/utils/file"
+	"fusenapi/utils/websocket_data"
+	"time"
+
+	"context"
+
+	"fusenapi/server/websocket/internal/svc"
+	"fusenapi/server/websocket/internal/types"
+
+	"github.com/zeromicro/go-zero/core/logx"
+)
+
+type RenderNotifyLogic struct {
+	logx.Logger
+	ctx    context.Context
+	svcCtx *svc.ServiceContext
+}
+
+func NewRenderNotifyLogic(ctx context.Context, svcCtx *svc.ServiceContext) *RenderNotifyLogic {
+	return &RenderNotifyLogic{
+		Logger: logx.WithContext(ctx),
+		ctx:    ctx,
+		svcCtx: svcCtx,
+	}
+}
+
+// 处理进入前逻辑w,r
+// func (l *RenderNotifyLogic) BeforeLogic(w http.ResponseWriter, r *http.Request) {
+// }
+
+// 处理逻辑后 w,r 如:重定向, resp 必须重新处理
+// func (l *RenderNotifyLogic) AfterLogic(w http.ResponseWriter, r *http.Request, resp *basic.Response) {
+// // httpx.OkJsonCtx(r.Context(), w, resp)
+// }
+
+func (l *RenderNotifyLogic) RenderNotify(req *types.RenderNotifyReq, userinfo *auth.UserInfo) (resp *basic.Response) {
+	if req.TaskId == "" {
+		return resp.SetStatusWithMessage(basic.CodeRequestParamsErr, "invalid param task_id")
+	}
+	if req.Image == "" {
+		return resp.SetStatusWithMessage(basic.CodeRequestParamsErr, "invalid param image")
+	}
+	if req.UserId == 0 && req.GuestId == 0 {
+		return resp.SetStatusWithMessage(basic.CodeRequestParamsErr, "invalid user_id or guest_id")
+	}
+	// 上传文件
+	var upload = file.Upload{
+		Ctx:        l.ctx,
+		MysqlConn:  l.svcCtx.MysqlConn,
+		AwsSession: l.svcCtx.AwsSession,
+	}
+	uploadRes, err := upload.UploadFileByBase64(&file.UploadBaseReq{
+		Source:       "unity cloud render",
+		FileHash:     req.TaskId,
+		FileData:     req.Image,
+		Metadata:     "",
+		UploadBucket: 1,
+		ApiType:      2,
+		UserId:       req.UserId,
+		GuestId:      req.GuestId,
+		FileByte:     nil,
+	})
+	if err != nil {
+		logx.Error(err)
+		return resp.SetStatusWithMessage(basic.CodeFileUploadErr, "failed to upload render resource image")
+	}
+	//遍历websocket链接把数据传进去
+	mapConnPool.Range(func(key, value any) bool {
+		//断言连接
+		ws, ok := value.(wsConnectItem)
+		if !ok {
+			return true
+		}
+		//关闭标识
+		if ws.isClose {
+			return true
+		}
+		//查询有无该渲染任务
+		renderId, ok := ws.renderProperty.renderImageTask[req.TaskId]
+		if !ok {
+			return true
+		}
+		b := ws.respondDataFormat(constants.WEBSOCKET_RENDER_IMAGE, websocket_data.RenderImageRspMsg{
+			RenderId: renderId,
+			Image:    uploadRes.ResourceUrl,
+		})
+		deleteTask := renderImageControlChanItem{
+			Option:   0, //0删除 1添加
+			TaskId:   req.TaskId,
+			RenderId: renderId,
+		}
+		select {
+		case <-ws.closeChan: //关闭了
+			return true
+		case ws.renderProperty.renderImageTaskCtlChan <- deleteTask: //删除对应的需要渲染的图片map
+			//发送数据到out chan
+			ws.sendToOutChan(b)
+		case <-time.After(time.Second * 3): //超时丢弃
+			return true
+		}
+		return true
+	})
+	logx.Info("渲染回调成功######################")
+	return resp.SetStatusWithMessage(basic.CodeOK, "success")
+}

server/websocket/internal/logic/ws_render_image_logic.go

@@ -1,20 +1,26 @@
 package logic
 
 import (
+	"bytes"
 	"encoding/json"
 	"errors"
 	"fusenapi/constants"
+	"fusenapi/service/repositories"
+	"fusenapi/utils/curl"
 	"fusenapi/utils/hash"
 	"fusenapi/utils/websocket_data"
 
 	"github.com/zeromicro/go-zero/core/logx"
 	"gorm.io/gorm"
+	"strconv"
+	"time"
 )
 
 // 云渲染属性
 type renderProperty struct {
 	renderImageTask        map[string]string               //需要渲染的图片任务 key是taskId val 是renderId
 	renderImageTaskCtlChan chan renderImageControlChanItem //渲染任务新增移除的控制通道
+	renderChan             chan []byte                     //渲染的缓冲队列
 }
 
 // 渲染任务新增移除的控制通道的数据
@@ -24,8 +30,39 @@ type renderImageControlChanItem struct {
 	RenderId string // map的val
 }
 
+// 发送到渲染缓冲池
+func (w *wsConnectItem) sendToRenderChan(data []byte) {
+	select {
+	case <-w.closeChan: //已经关闭
+		return
+	case w.renderProperty.renderChan <- data:
+		return
+	case <-time.After(time.Second * 3):
+		return
+	}
+}
+
 // 渲染发送到组装数据组装数据
-func (w *wsConnectItem) renderImage(data []byte) {
+func (w *wsConnectItem) renderImage() {
+	defer func() {
+		if err := recover(); err != nil {
+			logx.Error("renderImage panic:", err)
+		}
+	}()
+	for {
+		select {
+		case <-w.closeChan: //已关闭
+			return
+		case data := <-w.renderProperty.renderChan:
+			w.consumeRenderCache(data)
+		}
+	}
+
+}
+
+// 消费渲染缓冲数据
+func (w *wsConnectItem) consumeRenderCache(data []byte) {
+	logx.Info("消费渲染数据:", string(data))
 	var renderImageData websocket_data.RenderImageReqMsg
 	if err := json.Unmarshal(data, &renderImageData); err != nil {
 		w.sendToOutChan(w.respondDataFormat(constants.WEBSOCKET_ERR_DATA_FORMAT, "invalid format of websocket render image message:"+string(data)))
@@ -49,7 +86,7 @@ func (w *wsConnectItem) renderImage(data []byte) {
 		return
 	}
 	//获取上传最近的logo
-	userMaterial, err := w.allModels.FsUserMaterial.FindLatestOne(w.ctx, w.userId, w.guestId)
+	userMaterial, err := w.logic.svcCtx.AllModels.FsUserMaterial.FindLatestOne(w.logic.ctx, w.userId, w.guestId)
 	if err != nil {
 		if !errors.Is(err, gorm.ErrRecordNotFound) {
 			w.sendToOutChan(w.respondDataFormat(constants.WEBSOCKET_ERR_DATA_FORMAT, "failed to get user logo"))
@@ -57,7 +94,7 @@ func (w *wsConnectItem) renderImage(data []byte) {
 			return
 		}
 		//使用默认logo(id=0)
-		userMaterialDefault, err := w.allModels.FsUserMaterial.FindOneById(w.ctx, 0)
+		userMaterialDefault, err := w.logic.svcCtx.AllModels.FsUserMaterial.FindOneById(w.logic.ctx, 0)
 		if err != nil {
 			if errors.Is(err, gorm.ErrRecordNotFound) {
 				w.sendToOutChan(w.respondDataFormat(constants.WEBSOCKET_ERR_DATA_FORMAT, "default logo is not exists"))
@@ -75,10 +112,13 @@ func (w *wsConnectItem) renderImage(data []byte) {
 	renderImageData.RenderData.UserId = w.userId
 	renderImageData.RenderData.GuestId = w.guestId
 
-	//生成任务id
-	taskId := hash.JsonHashKey(renderImageData.RenderData)
+	//生成任务id(需要把user_id,guest_id设为0)
+	hashVal := renderImageData.RenderData
+	hashVal.UserId = 0
+	hashVal.GuestId = 0
+	taskId := hash.JsonHashKey(hashVal)
 	//查询有没有缓存的资源,有就返回######################
-	resource, err := w.allModels.FsResource.FindOneById(w.ctx, taskId)
+	resource, err := w.logic.svcCtx.AllModels.FsResource.FindOneById(w.logic.ctx, taskId)
 	if err != nil {
 		if !errors.Is(err, gorm.ErrRecordNotFound) {
 			logx.Error("failed to find render resource:", err)
@@ -101,18 +141,156 @@ func (w *wsConnectItem) renderImage(data []byte) {
 		TaskId:   taskId,
 		RenderId: renderImageData.RenderId,
 	}
-	tmpData := websocket_data.AssembleRenderData{
-		TaskId:     taskId,
-		RenderId:   renderImageData.RenderId,
-		RenderData: renderImageData.RenderData,
-	}
-	d, _ := json.Marshal(tmpData)
-	//发送给对应的流水线组装数据
-	if err = w.rabbitMq.SendMsg(constants.RABBIT_MQ_ASSEMBLE_RENDER_DATA, d); err != nil {
-		logx.Error("发送渲染任务数据到MQ失败:", string(d), "err:", err)
+	//组装数据
+	if err = w.assembleRenderData(taskId, renderImageData); err != nil {
+		logx.Error("组装数据失败:", err)
 		return
 	}
-	logx.Info("发送渲染数据到rabbitmq成功:", string(d))
+}
+
+// 组装数据发送给unity
+func (w *wsConnectItem) assembleRenderData(taskId string, info websocket_data.RenderImageReqMsg) error {
+	defer func() {
+		if err := recover(); err != nil {
+			logx.Error("assembleRenderData panic:", err)
+		}
+	}()
+	//获取模板
+	productTemplate, err := w.logic.svcCtx.AllModels.FsProductTemplateV2.FindOneByProductIdTagIdWithSizeTable(w.logic.ctx, info.RenderData.ProductId, info.RenderData.TemplateTag)
+	if err != nil {
+		if errors.Is(err, gorm.ErrRecordNotFound) {
+			logx.Error("template info is not found")
+			return err
+		}
+		logx.Error("failed to get template info:", err)
+		return err
+	}
+	//获取刀版图
+	res, err := w.logic.svcCtx.Repositories.ImageHandle.LogoCombine(w.logic.ctx, &repositories.LogoCombineReq{
+		UserId:      info.RenderData.UserId,
+		GuestId:     info.RenderData.GuestId,
+		TemplateId:  productTemplate.Id,
+		TemplateTag: info.RenderData.TemplateTag,
+		Website:     info.RenderData.Website,
+		Slogan:      info.RenderData.Slogan,
+		Address:     info.RenderData.Address,
+		Phone:       info.RenderData.Phone,
+	})
+	if err != nil {
+		logx.Error("合成刀版图失败:", err)
+		return err
+	}
+	combineImage := "" //刀版图
+	if res != nil && res.ResourceUrl != nil {
+		combineImage = *res.ResourceUrl
+	} else {
+		logx.Error("合成刀版图失败,合成的刀版图是空指针:", err)
+		return err
+	}
+	logx.Info("合成刀版图成功:", *res.ResourceUrl)
+	//获取渲染设置信息
+	element, err := w.logic.svcCtx.AllModels.FsProductTemplateElement.FindOneByModelId(w.logic.ctx, *productTemplate.ModelId)
+	if err != nil {
+		if errors.Is(err, gorm.ErrRecordNotFound) {
+			// todo 没有图就给他返回一张默认(后面要删除)
+			defaultImg := w.respondDataFormat(constants.WEBSOCKET_RENDER_IMAGE, websocket_data.RenderImageRspMsg{
+				RenderId: info.RenderId,
+				Image:    "https://s3.us-west-1.amazonaws.com/storage.fusenpack.com/695463af6e9b93c003db39ddf728241f9523efc55b20dc37f30fe5d96ed54fb5",
+			})
+			w.sendToOutChan(defaultImg)
+			logx.Error("element info is not found,model_id = ", *productTemplate.ModelId)
+			return err
+		}
+		logx.Error("failed to get element list,", err)
+		return err
+	}
+	//组装数据
+	refletion := -1
+	if element.Refletion != nil && *element.Refletion != "" {
+		refletion, err = strconv.Atoi(*element.Refletion)
+	}
+	//组装data数据
+	var mode map[string]interface{}
+	if element.Mode != nil && *element.Mode != "" {
+		if err = json.Unmarshal([]byte(*element.Mode), &mode); err != nil {
+			logx.Error("faile to parse element mode json:", err)
+			return err
+		}
+	}
+	tempData := make([]map[string]interface{}, 0, 3)
+	if element.Base != nil && *element.Base != "" {
+		tempData = append(tempData, map[string]interface{}{
+			"name":      "model",
+			"data":      "0," + combineImage + "," + *element.Base,
+			"type":      "other",
+			"layer":     "0",
+			"is_update": 1,
+			"mode":      mode["model"],
+		})
+	}
+	if element.Shadow != nil && *element.Shadow != "" {
+		tempData = append(tempData, map[string]interface{}{
+			"name":      "shadow",
+			"data":      *element.Shadow,
+			"type":      "other",
+			"layer":     "0",
+			"is_update": 0,
+			"mode":      mode["shadow"],
+		})
+	}
+	if element.ModelP != nil && *element.ModelP != "" {
+		tempData = append(tempData, map[string]interface{}{
+			"name":      "model_P",
+			"data":      "0," + *element.ModelP,
+			"type":      "other",
+			"layer":     "0",
+			"is_update": 0,
+			"mode":      mode["model_P"],
+		})
+	}
+	result := []interface{}{
+		map[string]interface{}{
+			"light":     *element.Light,
+			"refletion": refletion,
+			"scale":     *element.Scale,
+			"sku_id":    info.RenderData.ProductId,
+			"tid":       *element.Title,
+			"rotation":  *element.Rotation,
+			"filePath":  "", //todo 文件路径,针对千人千面
+			"data":      tempData,
+		},
+	}
+	sendData := map[string]interface{}{
+		"id":               taskId,
+		"order_id":         0,
+		"user_id":          info.RenderData.UserId,
+		"guest_id":         info.RenderData.GuestId,
+		"sku_ids":          []int64{info.RenderData.ProductId},
+		"tids":             []string{*element.Title},
+		"data":             result,
+		"is_thousand_face": 0,
+		"folder":           "", //todo 千人千面需要使用
+	}
+	//请求unity接口
+	url := "http://api.fusen.3718.cn:4050/api/render/queue/push"
+	header := make(map[string]string)
+	header["content-type"] = "application/json"
+	t := time.Now().UTC()
+	postData := map[string]interface{}{
+		"group":       "unity3d",
+		"source":      "home page",
+		"priority":    1,
+		"create_at":   t,
+		"render_data": sendData,
+	}
+	p, _ := json.Marshal(postData)
+	_, err = curl.ApiCall(url, "POST", header, bytes.NewReader(p), time.Second*10)
+	if err != nil {
+		logx.Error("failed to send data to unity")
+		return err
+	}
+	logx.Info("发送到unity成功################")
+	return nil
 }
 
 // 操作连接中渲染任务的增加/删除