package taskcenter import ( "context" "encoding/json" "errors" "fmt" "net/http" "strings" "time" "chatapp3-golang/internal/model" "chatapp3-golang/internal/utils" "gorm.io/gorm" ) // ReportSimpleEvent 供 Go 内部业务模块上报任务进度事件。 // 上报方只需要传标准 int64,内部再转换成兼容 HTTP 入参的 EventRequest。 func (s *Service) ReportSimpleEvent(ctx context.Context, sysOrigin, eventID, eventType string, userID, deltaValue int64, payload map[string]any) error { return s.ReportSimpleEventAt(ctx, sysOrigin, eventID, eventType, userID, deltaValue, time.Now(), payload) } // ReportSimpleEventAt 和 ReportSimpleEvent 相同,但允许 MQ 等异步事件传入真实发生时间。 func (s *Service) ReportSimpleEventAt(ctx context.Context, sysOrigin, eventID, eventType string, userID, deltaValue int64, occurredAt time.Time, payload map[string]any) error { _, err := s.ReceiveEvent(ctx, EventRequest{ SysOrigin: sysOrigin, EventID: eventID, EventType: eventType, UserID: flexibleInt64(userID), DeltaValue: deltaValue, OccurredAt: occurredAt.Format(time.RFC3339Nano), Payload: payload, }) if err != nil { return fmt.Errorf("report task event %s failed: %w", eventID, err) } return nil } // ReceiveEvent 接收后端业务事件并更新命中的任务进度。 func (s *Service) ReceiveEvent(ctx context.Context, req EventRequest) (*EventResponse, error) { sysOrigin := s.normalizeSysOrigin(req.SysOrigin) eventID := strings.TrimSpace(req.EventID) if eventID == "" { return nil, NewAppError(http.StatusBadRequest, "invalid_event_id", "eventId is required") } eventType := strings.ToUpper(strings.TrimSpace(req.EventType)) if eventType == "" { return nil, NewAppError(http.StatusBadRequest, "invalid_event_type", "eventType is required") } if !isAllowedConditionType(eventType) { return &EventResponse{Processed: false, Updated: 0}, nil } userID := req.UserID.Int64() if userID <= 0 { return nil, NewAppError(http.StatusBadRequest, "invalid_user_id", "userId is required") } if req.DeltaValue <= 0 { return nil, NewAppError(http.StatusBadRequest, "invalid_delta_value", "deltaValue must be greater than 0") } var existingCount int64 if err := s.db.WithContext(ctx).Model(&model.TaskCenterEvent{}). Where("sys_origin = ? AND event_id = ?", sysOrigin, eventID). Count(&existingCount).Error; err != nil { return nil, err } if existingCount > 0 { return &EventResponse{Processed: false, Updated: 0}, nil } bundle, err := s.loadEnabledBundle(ctx, sysOrigin) if err != nil { return nil, err } if bundle == nil { return &EventResponse{Processed: false, Updated: 0}, nil } matchedTasks := make([]model.TaskCenterTaskConfig, 0) for _, task := range bundle.Tasks { if strings.EqualFold(task.ConditionType, eventType) { matchedTasks = append(matchedTasks, task) } } if len(matchedTasks) == 0 { return &EventResponse{Processed: false, Updated: 0}, nil } occurredAt := parseOccurredAt(req.OccurredAt, time.Now()) rawJSON := "" if len(req.Payload) > 0 { if bytes, marshalErr := json.Marshal(req.Payload); marshalErr == nil { rawJSON = string(bytes) } } updated := 0 if err := s.db.WithContext(ctx).Transaction(func(tx *gorm.DB) error { var duplicateCount int64 if err := tx.Model(&model.TaskCenterEvent{}). Where("sys_origin = ? AND event_id = ?", sysOrigin, eventID). Count(&duplicateCount).Error; err != nil { return err } if duplicateCount > 0 { return nil } for _, task := range matchedTasks { cycleKey := currentCycleKey(task.TaskCategory, bundle.Location, occurredAt) progress, err := s.loadOrCreateProgressTx(tx, sysOrigin, userID, task, cycleKey) if err != nil { return err } if progress.RewardCollected { continue } progress.CompletedValue += req.DeltaValue if progress.CompletedValue >= progress.TargetValue { progress.Status = TaskStatusCompleted } else { progress.Status = TaskStatusInProgress } progress.UpdateTime = time.Now() if err := tx.Save(&progress).Error; err != nil { return err } eventRowID, idErr := utils.NextID() if idErr != nil { return idErr } eventRow := model.TaskCenterEvent{ ID: eventRowID, SysOrigin: sysOrigin, EventID: eventID, EventType: eventType, UserID: userID, TaskID: task.ID, TaskCode: task.TaskCode, TaskCategory: task.TaskCategory, CycleKey: cycleKey, DeltaValue: req.DeltaValue, CompletedValue: progress.CompletedValue, RawJSON: rawJSON, OccurredAt: occurredAt, CreateTime: time.Now(), UpdateTime: time.Now(), } if err := tx.Create(&eventRow).Error; err != nil { return err } updated++ } return nil }); err != nil { return nil, err } return &EventResponse{Processed: updated > 0, Updated: updated}, nil } func (s *Service) loadOrCreateProgressTx(tx *gorm.DB, sysOrigin string, userID int64, task model.TaskCenterTaskConfig, cycleKey string) (model.TaskCenterUserProgress, error) { var progress model.TaskCenterUserProgress err := tx.Where("sys_origin = ? AND user_id = ? AND task_id = ? AND cycle_key = ?", sysOrigin, userID, task.ID, cycleKey). First(&progress).Error if errors.Is(err, gorm.ErrRecordNotFound) { id, idErr := utils.NextID() if idErr != nil { return model.TaskCenterUserProgress{}, idErr } now := time.Now() progress = model.TaskCenterUserProgress{ ID: id, SysOrigin: sysOrigin, UserID: userID, TaskID: task.ID, TaskCode: task.TaskCode, TaskCategory: task.TaskCategory, CycleKey: cycleKey, ConditionType: task.ConditionType, CompletedValue: 0, TargetValue: task.TargetValue, Status: TaskStatusInProgress, CreateTime: now, UpdateTime: now, } return progress, nil } if err != nil { return model.TaskCenterUserProgress{}, err } if progress.TargetValue != task.TargetValue || progress.ConditionType != task.ConditionType { progress.TargetValue = task.TargetValue progress.ConditionType = task.ConditionType } return progress, nil }