bugfix
This commit is contained in:
+266
-101
@@ -3,8 +3,8 @@ package service
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"log/slog"
|
||||
"strings"
|
||||
"time"
|
||||
|
||||
"git.lzcwy.cn/zhengchiliu/zt-closet-backend/global"
|
||||
@@ -12,12 +12,24 @@ import (
|
||||
"git.lzcwy.cn/zhengchiliu/zt-closet-backend/internal/mysqldao"
|
||||
"git.lzcwy.cn/zhengchiliu/zt-closet-backend/internal/mysqlmodel"
|
||||
"git.lzcwy.cn/zhengchiliu/zt-closet-backend/internal/redisdao"
|
||||
"git.lzcwy.cn/zhengchiliu/zt-closet-backend/internal/redismodel"
|
||||
"git.lzcwy.cn/zhengchiliu/zt-closet-backend/pkg/errcode"
|
||||
"git.lzcwy.cn/zhengchiliu/zt-closet-backend/pkg/tokenhub"
|
||||
"gorm.io/gorm"
|
||||
)
|
||||
|
||||
const (
|
||||
// aiRecommendRunTimeout 后台生成任务的整体超时。
|
||||
// 大模型单次最长 120s,其余留给查库与落库,给足余量。
|
||||
aiRecommendRunTimeout = 3 * time.Minute
|
||||
|
||||
// aiRecommendSlotTTL 用户活跃位保留时长,需略长于任务超时,
|
||||
// 否则任务尚未收尾活跃位就先过期,会造成同一用户重复提交、重复扣额度。
|
||||
aiRecommendSlotTTL = aiRecommendRunTimeout + time.Minute
|
||||
|
||||
// aiRecommendCleanupTimeout 任务收尾(退额度 / 写流水 / 释放活跃位)的超时。
|
||||
aiRecommendCleanupTimeout = 15 * time.Second
|
||||
)
|
||||
|
||||
// AIRecommendService AI 穿搭推荐业务服务。
|
||||
type AIRecommendService struct {
|
||||
outfitDao *mysqldao.OutfitDao
|
||||
@@ -40,125 +52,99 @@ func NewAIRecommendService() *AIRecommendService {
|
||||
}
|
||||
}
|
||||
|
||||
// Create 发起 AI 穿搭推荐:防并发、扣 AI 额度、基于衣橱与运势生成推荐,落库为待保存穿搭。
|
||||
func (s *AIRecommendService) Create(ctx context.Context, userID int64, request *dto.AIRecommendCreateRequest) (*dto.AIRecommendCreateResponse, *errcode.Error) {
|
||||
ctx, span := global.Tracer.Start(ctx, "svc.AIRecommendCreate")
|
||||
// Submit 提交 AI 穿搭推荐任务,立即返回 task_id,真正的生成在后台完成。
|
||||
//
|
||||
// 与同步实现的区别:
|
||||
// - 接口不等待大模型,秒级返回,天然不受 Server.WriteTimeout 约束;
|
||||
// - 同一用户已有任务在执行时,直接返回该任务的 task_id(幂等),不重复扣额度;
|
||||
// - 生成过程使用独立 context,前端退出不会中断,结果仍会落库为「待保存」穿搭。
|
||||
func (s *AIRecommendService) Submit(ctx context.Context, userID int64, request *dto.AIRecommendCreateRequest) (*dto.AIRecommendSubmitResponse, *errcode.Error) {
|
||||
ctx, span := global.Tracer.Start(ctx, "svc.AIRecommendSubmit")
|
||||
defer span.End()
|
||||
|
||||
// 1. 防并发锁。
|
||||
lockKey := fmt.Sprintf(redismodel.KeyLockAIRecommend, userID)
|
||||
lockVal := genUUID()
|
||||
ok, err := redisdao.TryLock(ctx, lockKey, lockVal, 180*time.Second)
|
||||
taskID := genUUID()
|
||||
|
||||
// 1. 抢占活跃位:抢不到说明该用户已有任务在跑,幂等返回它。
|
||||
acquired, runningTaskID, err := redisdao.AIRecommendTaskAcquire(ctx, userID, taskID, aiRecommendSlotTTL)
|
||||
if err != nil {
|
||||
// Redis 异常:返回服务错误而非误报「请求过于频繁」。
|
||||
return nil, errcode.CacheError
|
||||
}
|
||||
if !ok {
|
||||
return nil, errcode.TooManyRequests
|
||||
if !acquired {
|
||||
if runningTaskID == "" {
|
||||
// 活跃位恰好在两次调用之间过期,让前端稍后重试即可。
|
||||
return nil, errcode.TooManyRequests
|
||||
}
|
||||
return &dto.AIRecommendSubmitResponse{
|
||||
TaskID: runningTaskID,
|
||||
Status: AIRecommendTaskPending,
|
||||
}, nil
|
||||
}
|
||||
defer func() { _ = redisdao.Unlock(ctx, lockKey, lockVal) }()
|
||||
|
||||
// 2. 扣减 AI 每日额度(权威扣减)。
|
||||
// 2. 扣减 AI 每日额度(权威扣减)。失败时释放活跃位,避免把用户卡死。
|
||||
if qerr := consumeDailyQuota(ctx, userID, QuotaTypeAI, 1, "AI 推荐消耗"); qerr != nil {
|
||||
s.releaseSlot(context.WithoutCancel(ctx), userID, taskID)
|
||||
if qerr == errcode.QuotaInsufficient {
|
||||
return nil, errcode.AIRecommendQuotaLimit
|
||||
}
|
||||
return nil, qerr
|
||||
}
|
||||
|
||||
// 失败时统一回退额度的兜底闭包。
|
||||
fail := func(bizErr *errcode.Error, reason string) *errcode.Error {
|
||||
refundDailyQuota(ctx, userID, QuotaTypeAI, 1)
|
||||
s.writeFlowLog(ctx, userID, "", AIFlowStatusFailed, reason)
|
||||
return bizErr
|
||||
// 3. 先落一个「生成中」快照,前端提交后立刻就能轮询到。
|
||||
if serr := redisdao.AIRecommendTaskSave(ctx, &redisdao.AIRecommendTask{
|
||||
TaskID: taskID,
|
||||
UserID: userID,
|
||||
Status: AIRecommendTaskPending,
|
||||
CreateTime: nowUnix(),
|
||||
}); serr != nil {
|
||||
cleanupCtx := context.WithoutCancel(ctx)
|
||||
refundDailyQuota(cleanupCtx, userID, QuotaTypeAI, 1)
|
||||
s.releaseSlot(cleanupCtx, userID, taskID)
|
||||
return nil, errcode.CacheError
|
||||
}
|
||||
|
||||
// 3. 加载用户可用衣物(已成功处理)。
|
||||
clothes, cerr := s.processedClothes(ctx, userID)
|
||||
if cerr != nil {
|
||||
return nil, fail(errcode.DatabaseError, "查询衣橱失败")
|
||||
}
|
||||
if len(clothes) == 0 {
|
||||
return nil, fail(errcode.AIRecommendFailed.WithDetails("衣橱暂无可用衣物"), "衣橱为空")
|
||||
}
|
||||
// 4. 后台生成。request 是栈上参数,复制一份交给 goroutine,避免外部复用导致数据竞争。
|
||||
reqCopy := *request
|
||||
go s.runTask(taskID, userID, &reqCopy)
|
||||
|
||||
// 4. 计算星座与今日运势(best-effort)。
|
||||
zodiac, fortune := s.userFortune(ctx, userID)
|
||||
|
||||
// 5. 调用大模型生成推荐。
|
||||
requestTime := nowUnix()
|
||||
client := tokenhub.NewClient()
|
||||
// 用户自定义场景(scene_type=2)透传给模型;为空时由模型自行判断。
|
||||
result, gerr := client.GenerateOutfitRecommendation(ctx, zodiac, fortune, clothes, request.Scene)
|
||||
if gerr != nil {
|
||||
return nil, fail(errcode.AIRecommendFailed, "大模型调用失败")
|
||||
}
|
||||
result.ValidateClothesIDs(clothes)
|
||||
if len(result.Items) == 0 {
|
||||
return nil, fail(errcode.AIRecommendFailed.WithDetails("未生成有效搭配"), "无有效单品")
|
||||
}
|
||||
|
||||
// 6. 落库为待用户保存的 AI 穿搭。
|
||||
scene := result.Scene
|
||||
if scene == "" {
|
||||
scene = request.Scene
|
||||
}
|
||||
outfit := &mysqlmodel.Outfit{
|
||||
OutfitUUID: genUUID(),
|
||||
UserID: userID,
|
||||
OutfitName: result.OutfitName,
|
||||
OutfitDesc: result.Description,
|
||||
Scene: scene,
|
||||
Source: OutfitSourceAI,
|
||||
IsPublic: OutfitPrivate,
|
||||
AuditStatus: OutfitAuditPending,
|
||||
Status: OutfitStatusUnsaved,
|
||||
}
|
||||
if err := global.MysqlEngine.WithContext(ctx).Transaction(func(tx *gorm.DB) error {
|
||||
if err := tx.Create(outfit).Error; err != nil {
|
||||
return err
|
||||
}
|
||||
clothesIDs := result.GetClothesIDs()
|
||||
items := make([]mysqlmodel.OutfitItem, 0, len(clothesIDs))
|
||||
for _, cid := range clothesIDs {
|
||||
items = append(items, mysqlmodel.OutfitItem{OutfitID: outfit.ID, ClothesID: cid})
|
||||
}
|
||||
return tx.Create(&items).Error
|
||||
}); err != nil {
|
||||
return nil, fail(errcode.OutfitCreateFailed, "穿搭及明细落库失败")
|
||||
}
|
||||
|
||||
// 7. 写成功流水 + 当日计数。
|
||||
resultJSON, _ := json.Marshal(result)
|
||||
if err := s.flowLogDao.AIRecommendFlowLogCreate(ctx, &mysqlmodel.AIRecommendFlowLog{
|
||||
RequestID: genUUID(),
|
||||
UserID: userID,
|
||||
ModelVersion: global.TokenhubSetting.Model,
|
||||
ResultContent: string(resultJSON),
|
||||
Status: AIFlowStatusSuccess,
|
||||
RequestTime: requestTime,
|
||||
ResponseTime: nowUnix(),
|
||||
}); err != nil {
|
||||
slog.WarnContext(ctx, "write ai recommend success flow log failed", "user_id", userID, "error", err)
|
||||
}
|
||||
|
||||
// 8. 组装响应(衣物 UUID 来自校正后的 items)。
|
||||
outfitItems := make([]dto.OutfitItem, 0, len(result.Items))
|
||||
for _, it := range result.Items {
|
||||
outfitItems = append(outfitItems, dto.OutfitItem{ClothesUUID: it.ClothesUUID})
|
||||
}
|
||||
|
||||
return &dto.AIRecommendCreateResponse{
|
||||
OutfitUUID: outfit.OutfitUUID,
|
||||
OutfitName: outfit.OutfitName,
|
||||
OutfitDesc: outfit.OutfitDesc,
|
||||
Scene: outfit.Scene,
|
||||
Source: outfit.Source,
|
||||
Status: outfit.Status,
|
||||
OutfitItem: outfitItems,
|
||||
return &dto.AIRecommendSubmitResponse{
|
||||
TaskID: taskID,
|
||||
Status: AIRecommendTaskPending,
|
||||
}, nil
|
||||
}
|
||||
|
||||
// Status 查询 AI 推荐任务的状态与结果。
|
||||
func (s *AIRecommendService) Status(ctx context.Context, userID int64, request *dto.AIRecommendStatusRequest) (*dto.AIRecommendStatusResponse, *errcode.Error) {
|
||||
ctx, span := global.Tracer.Start(ctx, "svc.AIRecommendStatus")
|
||||
defer span.End()
|
||||
|
||||
task, err := redisdao.AIRecommendTaskGet(ctx, request.TaskID)
|
||||
if err != nil {
|
||||
return nil, errcode.CacheError
|
||||
}
|
||||
// 任务不存在、已过期,或不属于当前用户时统一按「不存在」处理,避免探测他人任务。
|
||||
if task == nil || task.UserID != userID {
|
||||
return nil, errcode.AIRecommendTaskNotFound
|
||||
}
|
||||
|
||||
resp := &dto.AIRecommendStatusResponse{
|
||||
TaskID: task.TaskID,
|
||||
Status: task.Status,
|
||||
ErrorCode: task.ErrorCode,
|
||||
ErrorMsg: task.ErrorMsg,
|
||||
}
|
||||
|
||||
if task.Status == AIRecommendTaskSuccess && len(task.Result) > 0 {
|
||||
var out dto.AIRecommendCreateResponse
|
||||
if uerr := json.Unmarshal(task.Result, &out); uerr != nil {
|
||||
slog.ErrorContext(ctx, "unmarshal ai recommend task result failed",
|
||||
"task_id", task.TaskID, "user_id", userID, "error", uerr)
|
||||
return nil, errcode.AIRecommendFailed
|
||||
}
|
||||
resp.Result = &out
|
||||
}
|
||||
return resp, nil
|
||||
}
|
||||
|
||||
// Save 将 AI 推荐生成的待保存穿搭保存到「我的搭配」(置为正常状态)。
|
||||
func (s *AIRecommendService) Save(ctx context.Context, userID int64, request *dto.AIRecommendSaveRequest) *errcode.Error {
|
||||
ctx, span := global.Tracer.Start(ctx, "svc.AIRecommendSave")
|
||||
@@ -190,6 +176,166 @@ func (s *AIRecommendService) Save(ctx context.Context, userID int64, request *dt
|
||||
return nil
|
||||
}
|
||||
|
||||
// ---- 后台任务 ----
|
||||
|
||||
// runTask 后台执行推荐生成并回写任务状态。
|
||||
//
|
||||
// 关键点:ctx 由 context.Background() 派生,与 HTTP 请求生命周期彻底解耦 ——
|
||||
// 前端退出后生成仍会跑完并落库,额度退还、流水写入与活跃位释放也能正常收尾。
|
||||
func (s *AIRecommendService) runTask(taskID string, userID int64, request *dto.AIRecommendCreateRequest) {
|
||||
ctx, cancel := context.WithTimeout(context.Background(), aiRecommendRunTimeout)
|
||||
defer cancel()
|
||||
|
||||
defer func() {
|
||||
if r := recover(); r != nil {
|
||||
slog.ErrorContext(ctx, "ai recommend task panic",
|
||||
"task_id", taskID, "user_id", userID, "recover", r)
|
||||
s.finishFailed(ctx, taskID, userID, errcode.AIRecommendFailed.Code(), "生成任务异常终止")
|
||||
}
|
||||
// 无论成功、失败还是 panic,都要释放活跃位。
|
||||
s.releaseSlot(context.WithoutCancel(ctx), userID, taskID)
|
||||
}()
|
||||
|
||||
out, bizErr := s.generate(ctx, userID, request)
|
||||
if bizErr != nil {
|
||||
s.finishFailed(ctx, taskID, userID, bizErr.Code(), formatBizErr(bizErr))
|
||||
return
|
||||
}
|
||||
|
||||
if serr := redisdao.AIRecommendTaskSave(ctx, &redisdao.AIRecommendTask{
|
||||
TaskID: taskID,
|
||||
UserID: userID,
|
||||
Status: AIRecommendTaskSuccess,
|
||||
Result: mustMarshal(out),
|
||||
CreateTime: nowUnix(),
|
||||
}); serr != nil {
|
||||
// 结果已落库为主,快照写失败只影响前端轮询展示,记录即可。
|
||||
slog.ErrorContext(ctx, "save ai recommend task result failed",
|
||||
"task_id", taskID, "user_id", userID, "error", serr)
|
||||
}
|
||||
}
|
||||
|
||||
// finishFailed 任务失败的统一收尾:退额度、写失败流水、回写任务状态。
|
||||
// 收尾用的 ctx 不受任务取消影响(任务超时时 ctx 已 Done),否则会出现「额度不退、流水不写」。
|
||||
func (s *AIRecommendService) finishFailed(ctx context.Context, taskID string, userID int64, code int, msg string) {
|
||||
cleanupCtx, cancel := context.WithTimeout(context.WithoutCancel(ctx), aiRecommendCleanupTimeout)
|
||||
defer cancel()
|
||||
|
||||
refundDailyQuota(cleanupCtx, userID, QuotaTypeAI, 1)
|
||||
s.writeFlowLog(cleanupCtx, userID, "", AIFlowStatusFailed, msg)
|
||||
|
||||
if err := redisdao.AIRecommendTaskSave(cleanupCtx, &redisdao.AIRecommendTask{
|
||||
TaskID: taskID,
|
||||
UserID: userID,
|
||||
Status: AIRecommendTaskFailed,
|
||||
ErrorCode: code,
|
||||
ErrorMsg: msg,
|
||||
CreateTime: nowUnix(),
|
||||
}); err != nil {
|
||||
slog.ErrorContext(cleanupCtx, "save failed ai recommend task failed",
|
||||
"task_id", taskID, "user_id", userID, "error", err)
|
||||
}
|
||||
}
|
||||
|
||||
// releaseSlot 释放用户活跃位(best-effort,失败仅告警)。
|
||||
func (s *AIRecommendService) releaseSlot(ctx context.Context, userID int64, taskID string) {
|
||||
relCtx, cancel := context.WithTimeout(ctx, aiRecommendCleanupTimeout)
|
||||
defer cancel()
|
||||
|
||||
if err := redisdao.AIRecommendTaskRelease(relCtx, userID, taskID); err != nil {
|
||||
slog.WarnContext(relCtx, "release ai recommend slot failed",
|
||||
"user_id", userID, "task_id", taskID, "error", err)
|
||||
}
|
||||
}
|
||||
|
||||
// generate 执行一次完整的推荐生成:查衣橱 → 取运势 → 调大模型 → 校验 → 落库。
|
||||
func (s *AIRecommendService) generate(ctx context.Context, userID int64, request *dto.AIRecommendCreateRequest) (*dto.AIRecommendCreateResponse, *errcode.Error) {
|
||||
// 1. 加载用户可用衣物(已成功处理)。
|
||||
clothes, cerr := s.processedClothes(ctx, userID)
|
||||
if cerr != nil {
|
||||
return nil, errcode.DatabaseError
|
||||
}
|
||||
if len(clothes) == 0 {
|
||||
return nil, errcode.AIRecommendFailed.WithDetails("衣橱暂无可用衣物")
|
||||
}
|
||||
|
||||
// 2. 计算星座与今日运势(best-effort)。
|
||||
zodiac, fortune := s.userFortune(ctx, userID)
|
||||
|
||||
// 3. 调用大模型生成推荐。
|
||||
requestTime := nowUnix()
|
||||
client := tokenhub.NewClient()
|
||||
result, gerr := client.GenerateOutfitRecommendation(ctx, zodiac, fortune, clothes, request.Scene)
|
||||
if gerr != nil {
|
||||
slog.ErrorContext(ctx, "generate outfit recommendation failed", "user_id", userID, "error", gerr)
|
||||
return nil, errcode.AIRecommendFailed.WithDetails("大模型调用失败")
|
||||
}
|
||||
result.ValidateClothesIDs(clothes)
|
||||
if len(result.Items) == 0 {
|
||||
return nil, errcode.AIRecommendFailed.WithDetails("未生成有效搭配")
|
||||
}
|
||||
|
||||
// 4. 落库为待用户保存的 AI 穿搭。
|
||||
scene := result.Scene
|
||||
if scene == "" {
|
||||
scene = request.Scene
|
||||
}
|
||||
outfit := &mysqlmodel.Outfit{
|
||||
OutfitUUID: genUUID(),
|
||||
UserID: userID,
|
||||
OutfitName: result.OutfitName,
|
||||
OutfitDesc: result.Description,
|
||||
Scene: scene,
|
||||
Source: OutfitSourceAI,
|
||||
IsPublic: OutfitPrivate,
|
||||
AuditStatus: OutfitAuditPending,
|
||||
Status: OutfitStatusUnsaved,
|
||||
}
|
||||
if err := global.MysqlEngine.WithContext(ctx).Transaction(func(tx *gorm.DB) error {
|
||||
if err := tx.Create(outfit).Error; err != nil {
|
||||
return err
|
||||
}
|
||||
clothesIDs := result.GetClothesIDs()
|
||||
items := make([]mysqlmodel.OutfitItem, 0, len(clothesIDs))
|
||||
for _, cid := range clothesIDs {
|
||||
items = append(items, mysqlmodel.OutfitItem{OutfitID: outfit.ID, ClothesID: cid})
|
||||
}
|
||||
return tx.Create(&items).Error
|
||||
}); err != nil {
|
||||
return nil, errcode.OutfitCreateFailed
|
||||
}
|
||||
|
||||
// 5. 写成功流水(best-effort)。
|
||||
resultJSON, _ := json.Marshal(result)
|
||||
if err := s.flowLogDao.AIRecommendFlowLogCreate(ctx, &mysqlmodel.AIRecommendFlowLog{
|
||||
RequestID: genUUID(),
|
||||
UserID: userID,
|
||||
ModelVersion: global.TokenhubSetting.Model,
|
||||
ResultContent: string(resultJSON),
|
||||
Status: AIFlowStatusSuccess,
|
||||
RequestTime: requestTime,
|
||||
ResponseTime: nowUnix(),
|
||||
}); err != nil {
|
||||
slog.WarnContext(ctx, "write ai recommend success flow log failed", "user_id", userID, "error", err)
|
||||
}
|
||||
|
||||
// 6. 组装响应(衣物 UUID 来自校正后的 items)。
|
||||
outfitItems := make([]dto.OutfitItem, 0, len(result.Items))
|
||||
for _, it := range result.Items {
|
||||
outfitItems = append(outfitItems, dto.OutfitItem{ClothesUUID: it.ClothesUUID})
|
||||
}
|
||||
|
||||
return &dto.AIRecommendCreateResponse{
|
||||
OutfitUUID: outfit.OutfitUUID,
|
||||
OutfitName: outfit.OutfitName,
|
||||
OutfitDesc: outfit.OutfitDesc,
|
||||
Scene: outfit.Scene,
|
||||
Source: outfit.Source,
|
||||
Status: outfit.Status,
|
||||
OutfitItem: outfitItems,
|
||||
}, nil
|
||||
}
|
||||
|
||||
// ---- 内部辅助 ----
|
||||
|
||||
// processedClothes 获取用户已成功处理、可用于推荐的衣物。
|
||||
@@ -254,3 +400,22 @@ func (s *AIRecommendService) writeFlowLog(ctx context.Context, userID int64, res
|
||||
slog.WarnContext(ctx, "write ai recommend flow log failed", "user_id", userID, "status", status, "error", err)
|
||||
}
|
||||
}
|
||||
|
||||
// formatBizErr 拼接业务错误主文案与明细,便于前端展示失败原因。
|
||||
func formatBizErr(e *errcode.Error) string {
|
||||
msg := e.Msg()
|
||||
if details := e.Details(); len(details) > 0 {
|
||||
msg = msg + ":" + strings.Join(details, ";")
|
||||
}
|
||||
return msg
|
||||
}
|
||||
|
||||
// mustMarshal 序列化任务结果;失败时返回 nil(任务仍标记成功,前端可去「我的搭配」查看)。
|
||||
func mustMarshal(v any) json.RawMessage {
|
||||
data, err := json.Marshal(v)
|
||||
if err != nil {
|
||||
slog.Error("marshal ai recommend task result failed", "error", err)
|
||||
return nil
|
||||
}
|
||||
return data
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user