fusen/fusenapi
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

fix:上下文

Browse Source
This commit is contained in:
momo
2023-08-25 16:53:33 +08:00
parent bb77d8f9aa 7a02b3242e
commit 5cba35bdc7
30 changed files with 537 additions and 61 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
server/auth/internal/logic/acceptcookielogic.go
View File

@@ -44,7 +44,7 @@ func (l *AcceptCookieLogic) AcceptCookie(req *types.Request, userinfo *auth.User
}
m := l.svcCtx.AllModels.FsGuest
token, err := m.GenerateGuestID(l.ctx, &l.svcCtx.Config.Auth.AccessSecret)
token, err := m.GenerateGuestID(l.ctx, auth.DefaultJwtSecret)
if err != nil {
return resp.SetStatus(basic.CodeGuestGenErr)
}

6
server/auth/internal/logic/email_manager.go
View File

@@ -54,7 +54,7 @@ func (m *EmailSender) ProcessEmailTasks() {
m.emailSending[emailformat.TargetEmail] = &EmailTask{
Email: emailformat,
SendTime: time.Now(),
SendTime: time.Now().UTC(),
}
m.lock.Unlock()
@@ -82,7 +82,7 @@ func (m *EmailSender) Resend(emailTarget string, content []byte) {
defer m.lock.Unlock()
// Check if the email task still exists and has not been sent successfully
if task, ok := m.emailSending[emailTarget]; ok && task.SendTime.Add(m.ResendTimeLimit).After(time.Now()) {
if task, ok := m.emailSending[emailTarget]; ok && task.SendTime.Add(m.ResendTimeLimit).After(time.Now().UTC()) {
err := smtp.SendMail(emailTarget, m.Auth, m.FromEmail, []string{emailTarget}, content)
if err != nil {
log.Printf("Failed to resend email to %s: %v\n", emailTarget, err)
@@ -102,7 +102,7 @@ func (m *EmailSender) ClearExpiredTasks() {
m.lock.Lock()
for email, task := range m.emailSending {
if task.SendTime.Add(m.ResendTimeLimit).Before(time.Now()) {
if task.SendTime.Add(m.ResendTimeLimit).Before(time.Now().UTC()) {
delete(m.emailSending, email)
}
}

70
server/auth/internal/logic/useremailconfirmationlogic.go
View File

@@ -1,9 +1,10 @@
package logic
import (
"fmt"
"fusenapi/utils/auth"
"fusenapi/utils/basic"
"log"
"fusenapi/utils/wevent"
"time"
"context"
@@ -11,6 +12,7 @@ import (
"fusenapi/server/auth/internal/svc"
"fusenapi/server/auth/internal/types"
"github.com/474420502/requests"
"github.com/zeromicro/go-zero/core/logx"
)
@@ -53,7 +55,7 @@ func (l *UserEmailConfirmationLogic) UserEmailConfirmation(req *types.RequestEma
// 谷歌平台的注册流程
user, err := l.svcCtx.AllModels.FsUser.RegisterByGoogleOAuth(l.ctx, token)
if err != nil {
logx.Error(err)
logx.Error(err, token.TraceId)
return resp.SetStatus(basic.CodeDbSqlErr)
}
@@ -67,15 +69,75 @@ func (l *UserEmailConfirmationLogic) UserEmailConfirmation(req *types.RequestEma
)
if err != nil {
logx.Error(err)
logx.Error(err, token.TraceId)
return
}
log.Println(jwtToken) // 通过websocket去, 送回通道
event := wevent.NewWebsocketEventSuccess(wevent.UserEmailRegister, token.TraceId)
event.Data = wevent.DataEmailRegister{
JwtToken: jwtToken,
}
tp := requests.Post(fmt.Sprintf("%s/api/websocket/common_notify", l.svcCtx.Config.MainAddress))
tp.SetBodyJson(requests.M{
"wid": token.Wid,
"data": event,
})
wresp, err := tp.Execute()
if err != nil {
logx.Error(err, token.TraceId)
return
}
result := wresp.Json()
if result.Get("code").Int() != 200 {
logx.Error(result.Get("message"))
return
}
logx.Info("success", token.TraceId, jwtToken)
case "facebook":
case "fusen":
user, err := l.svcCtx.AllModels.FsUser.RegisterByFusen(l.ctx, token)
if err != nil {
logx.Error(err, token.TraceId)
return resp.SetStatus(basic.CodeDbSqlErr)
}
// 创建签证
jwtToken, err := auth.GenerateJwtTokenUint64(
auth.StringToHash(*user.PasswordHash),
l.svcCtx.Config.Auth.AccessExpire,
time.Now().Unix(),
user.Id,
0,
)
if err != nil {
logx.Error(err, token.TraceId)
return
}
event := wevent.NewWebsocketEventSuccess(wevent.UserEmailRegister, token.TraceId)
event.Data = wevent.DataEmailRegister{
JwtToken: jwtToken,
}
tp := requests.Post(fmt.Sprintf("%s/api/websocket/common_notify", l.svcCtx.Config.MainAddress))
tp.SetBodyJson(requests.M{
"wid": token.Wid,
"data": event,
})
wresp, err := tp.Execute()
if err != nil {
logx.Error(err, token.TraceId)
}
result := wresp.Json()
if result.Get("code").Int() != 200 {
logx.Error(result.Get("message"))
}
logx.Info("success", token.TraceId, jwtToken)
}
default:

4
server/auth/internal/logic/usergoogleloginlogic.go
View File

@@ -17,6 +17,7 @@ import (
"fusenapi/server/auth/internal/types"
"github.com/474420502/requests"
"github.com/google/uuid"
"github.com/zeromicro/go-zero/core/logx"
"github.com/zeromicro/go-zero/rest/httpx"
"golang.org/x/oauth2"
@@ -91,9 +92,10 @@ func (l *UserGoogleLoginLogic) UserGoogleLogin(req *types.RequestGoogleLogin, us
l.registerInfo = &auth.RegisterToken{
Id: googleId,
Password: base64.URLEncoding.EncodeToString(nonce),
Password: base64.RawURLEncoding.EncodeToString(nonce),
Platform: "google",
OperateType: auth.OpTypeRegister,
TraceId: uuid.NewString(),
CreateAt: time.Now(),
}

2
server/auth/internal/logic/userregisterlogic.go
View File

@@ -10,6 +10,7 @@ import (
"fusenapi/server/auth/internal/svc"
"fusenapi/server/auth/internal/types"
"github.com/google/uuid"
"github.com/zeromicro/go-zero/core/logx"
)
@@ -48,6 +49,7 @@ func (l *UserRegisterLogic) UserRegister(req *types.RequestUserRegister, userinf
Email: req.Email,
Password: req.Password,
Platform: "fusen",
TraceId: uuid.NewString(),
CreateAt: time.Now(),
}

2
server/auth/internal/logic/userresettokenlogic.go
View File

@@ -10,6 +10,7 @@ import (
"fusenapi/server/auth/internal/svc"
"fusenapi/server/auth/internal/types"
"github.com/google/uuid"
"github.com/zeromicro/go-zero/core/logx"
)
@@ -51,6 +52,7 @@ func (l *UserResetTokenLogic) UserResetToken(req *types.RequestUserResetToken, u
Wid: req.Wid,
Email: *user.Email,
OldPassword: *user.PasswordHash,
TraceId: uuid.NewString(),
CreateAt: time.Now(),
}

40
server/auth/internal/logic/websocket_test.go Normal file
View File

@@ -0,0 +1,40 @@
package logic
import (
"fmt"
"fusenapi/utils/wevent"
"log"
"testing"
"github.com/474420502/requests"
"github.com/google/uuid"
"github.com/zeromicro/go-zero/core/logx"
)
func TestSender(t *testing.T) {
traceId := uuid.NewString()
event := wevent.NewWebsocketEventSuccess(wevent.UserEmailRegister, traceId)
event.Data = wevent.DataEmailRegister{
JwtToken: traceId,
}
tp := requests.Post(fmt.Sprintf("%s/api/websocket/common_notify", "https://server.fusen.3718.cn:9900"))
tp.SetBodyJson(requests.M{
"wid": "tGyMYX9EldtsPLZTyT6PxrRgEV615CQGEiu9Sb1XrjZ4kpTjI46sQyh7kYfVlgN9uR5Uw4KDF+S62IknmaRgSMdee1QHVtCv+VEKrMF76snR04zS1ZbWZCgX5Lv2xgHz/bZBWwJF/9u6YTy2/FetGg==",
"data": event,
})
wresp, err := tp.Execute()
if err != nil {
logx.Error(err, "traceId")
return
}
result := wresp.Json()
if result.Get("code").Int() != 200 {
logx.Error(result.Get("message"))
return
}
log.Println(result)
}

76
server/websocket/internal/logic/commonnotifylogic.go
View File

@@ -5,6 +5,8 @@ import (
"fusenapi/constants"
"fusenapi/utils/auth"
"fusenapi/utils/basic"
"sync"
"time"
"fusenapi/server/websocket/internal/svc"
"fusenapi/server/websocket/internal/types"
@@ -26,6 +28,68 @@ func NewCommonNotifyLogic(ctx context.Context, svcCtx *svc.ServiceContext) *Comm
}
}
// 定义公共回调未找到websocket连接时暂存数据缓冲队列
var commonConnectionNotFoundDataCacheChan = make(chan commonConnectionNotFoundDataCacheChanItem, 2000)
type commonConnectionNotFoundDataCacheChanItem struct {
retryTimes int //重回队列次数
data types.CommonNotifyReq //数据
}
// 放入缓冲队列
func (l *CommonNotifyLogic) pushCommonCache(data commonConnectionNotFoundDataCacheChanItem) {
select {
case commonConnectionNotFoundDataCacheChan <- data:
return
case <-time.After(time.Millisecond * 50): //超50ms就丢弃
return
}
}
// 取出元素
func (l *CommonNotifyLogic) popCommonCache() (data commonConnectionNotFoundDataCacheChanItem) {
return <-commonConnectionNotFoundDataCacheChan
}
// 保证处理消息就一个循环在执行
var consumeCommonCacheData sync.Once
// 消费公共通知未处理的消息
func (l *CommonNotifyLogic) consumeCommonCacheData() {
//单例
consumeCommonCacheData.Do(func() {
tick := time.Tick(time.Millisecond * 200)
for {
select {
case <-tick: //200毫秒触发一次
info := l.popCommonCache()
//查询websocket连接
value, ok := mapConnPool.Load(info.data.Wid)
//没有连接
if !ok {
info.retryTimes--
//大于0，则放回队列
if info.retryTimes > 0 {
l.pushCommonCache(info)
continue
}
//否则直接丢弃消息
continue
}
//断言连接
ws, ok := value.(wsConnectItem)
if !ok {
logx.Error("渲染回调断言websocket连接失败")
continue
}
//发送
ws.sendToOutChan(ws.respondDataFormat(constants.WEBSOCKET_COMMON_NOTIFY, info.data.Data))
}
}
})
}
// 处理进入前逻辑w,r
// func (l *CommonNotifyLogic) BeforeLogic(w http.ResponseWriter, r *http.Request) {
// }
@@ -35,10 +99,20 @@ func (l *CommonNotifyLogic) CommonNotify(req *types.CommonNotifyReq, userinfo *a
if req.Wid == "" {
return resp.SetStatusWithMessage(basic.CodeRequestParamsErr, "websocket connect id is empty")
}
//触发消费公共未处理的消息（该方法是单例）
go l.consumeCommonCacheData()
//查询websocket连接
value, ok := mapConnPool.Load(req.Wid)
if !ok {
return resp.SetStatusWithMessage(basic.CodeOK, "success,but connection is not found")
//没找到连接就放到公共缓冲队列
go l.pushCommonCache(commonConnectionNotFoundDataCacheChanItem{
retryTimes: 20, //重试20次
data: types.CommonNotifyReq{
Wid: req.Wid,
Data: req.Data,
},
})
return resp.SetStatusWithMessage(basic.CodeOK, "success")
}
//断言连接
ws, ok := value.(wsConnectItem)

14
server/websocket/internal/logic/datatransferlogic.go
View File

@@ -5,6 +5,7 @@ import (
"bytes"
"encoding/hex"
"encoding/json"
"errors"
"fusenapi/constants"
"fusenapi/utils/auth"
"fusenapi/utils/encryption_decryption"
@@ -138,8 +139,8 @@ func (l *DataTransferLogic) DataTransfer(w http.ResponseWriter, r *http.Request)
// 设置连接
func (l *DataTransferLogic) setConnPool(conn *websocket.Conn, userInfo auth.UserInfo, isFirefoxBrowser bool) (wsConnectItem, error) {
//生成连接唯一标识
uniqueId, err := l.getUniqueId(userInfo)
//生成连接唯一标识（失败重试10次）
uniqueId, err := l.getUniqueId(userInfo, 10)
if err != nil {
//发送获取唯一标识失败的消息
l.sendGetUniqueIdErrResponse(conn)
@@ -170,11 +171,15 @@ func (l *DataTransferLogic) setConnPool(conn *websocket.Conn, userInfo auth.User
}
// 获取唯一id
func (l *DataTransferLogic) getUniqueId(userInfo auth.UserInfo) (uniqueId string, err error) {
func (l *DataTransferLogic) getUniqueId(userInfo auth.UserInfo, retryTimes int) (uniqueId string, err error) {
if retryTimes < 0 {
return "", errors.New("failed to get unique id")
}
//后面拼接上用户id
uniqueId = hex.EncodeToString([]byte(uuid.New().String())) + getUserJoinPart(userInfo.UserId, userInfo.GuestId)
//存在则从新获取
if _, ok := mapConnPool.Load(uniqueId); ok {
uniqueId, err = l.getUniqueId(userInfo)
uniqueId, err = l.getUniqueId(userInfo, retryTimes-1)
if err != nil {
return "", err
}
@@ -204,6 +209,7 @@ func (l *DataTransferLogic) checkAuth(r *http.Request) (isAuth bool, userInfo *a
logx.Error(err)
return false, nil
}
//todo 对接登录后要删除
userInfo.UserId = 39
userInfo.GuestId = 0
return true, userInfo

28
server/websocket/internal/logic/ws_render_image_logic.go → server/websocket/internal/logic/ws_render_image.go
View File

@@ -165,10 +165,14 @@ func (w *wsConnectItem) consumeRenderCache(data []byte) {
} else {
//返回给客户端
b := w.respondDataFormat(constants.WEBSOCKET_RENDER_IMAGE, websocket_data.RenderImageRspMsg{
RenderId: renderImageData.RenderId,
Image: *resource.ResourceUrl,
CombineTakesTime: "cache",
UnityRenderTakesTime: "cache",
RenderId: renderImageData.RenderId,
Image: *resource.ResourceUrl,
RenderProcessTime: websocket_data.RenderProcessTime{
CombineTakesTime: "cache",
UnityRenderTakesTime: "cache",
UploadCombineImageTakesTime: "cache",
UploadUnityRenderImageTakesTime: "cache",
},
})
//发送数据到out chan
w.sendToOutChan(b)
@@ -228,7 +232,7 @@ func (w *wsConnectItem) assembleRenderData(taskId string, info websocket_data.Re
}
res, err := w.logic.svcCtx.Repositories.ImageHandle.LogoCombine(w.logic.ctx, &combineReq)
if err != nil {
w.renderErrResponse(info.RenderId, info.RenderData.TemplateTag, taskId, "failed to combine image", w.userId, w.guestId, productTemplate.Id, model3dInfo.Id, productFirstSize.Id)
w.renderErrResponse(info.RenderId, info.RenderData.TemplateTag, taskId, "failed to combine image:"+err.Error(), w.userId, w.guestId, productTemplate.Id, model3dInfo.Id, productFirstSize.Id)
logx.Error("合成刀版图失败,合成请求数据:", combineReq, "错误信息:", err)
return err
}
@@ -438,12 +442,14 @@ func (w *wsConnectItem) operationRenderTask() {
}
//发送到出口
w.sendToOutChan(w.respondDataFormat(constants.WEBSOCKET_RENDER_IMAGE, websocket_data.RenderImageRspMsg{
RenderId: taskData.RenderId,
Image: data.RenderNotifyImageUrl,
CombineTakesTime: CombineTakesTime,
UnityRenderTakesTime: UnityRenderTakesTime,
UploadCombineImageTakesTime: uploadCombineImageTakesTime,
UploadUnityRenderImageTakesTime: uploadUnityRenderImageTakesTime,
RenderId: taskData.RenderId,
Image: data.RenderNotifyImageUrl,
RenderProcessTime: websocket_data.RenderProcessTime{
CombineTakesTime: CombineTakesTime,
UnityRenderTakesTime: UnityRenderTakesTime,
UploadCombineImageTakesTime: uploadCombineImageTakesTime,
UploadUnityRenderImageTakesTime: uploadUnityRenderImageTakesTime,
},
}))
}
//删除任务

1
server/websocket/internal/logic/ws_reuse_last_connect.go
View File

@@ -53,6 +53,7 @@ func (w *wsConnectItem) reuseLastConnect(data []byte) {
}
//重新绑定
w.uniqueId = wid
mapConnPool.Store(wid, *w)
rsp := w.respondDataFormat(constants.WEBSOCKET_CONNECT_SUCCESS, wid)
w.sendToOutChan(rsp)
return