278 lines
8.5 KiB
Go
278 lines
8.5 KiB
Go
package taskcenter
|
||
|
||
import (
|
||
"context"
|
||
"encoding/json"
|
||
"errors"
|
||
"fmt"
|
||
"net/http"
|
||
"strings"
|
||
"time"
|
||
|
||
"chatapp3-golang/internal/model"
|
||
"chatapp3-golang/internal/utils"
|
||
|
||
"gorm.io/gorm"
|
||
"gorm.io/gorm/clause"
|
||
)
|
||
|
||
// 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.ProcessEvent(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) {
|
||
return s.ProcessEvent(ctx, req)
|
||
}
|
||
|
||
// ProcessEvent 统一处理 HTTP 和 MQ 上报的任务事件。
|
||
func (s *Service) ProcessEvent(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, Reason: "unsupported_condition_type"}, 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")
|
||
}
|
||
occurredAt := parseOccurredAt(req.OccurredAt, time.Now())
|
||
receivedAt := time.Now()
|
||
rawJSON := buildArchiveRawJSON(req, sysOrigin, eventID, eventType, userID, req.DeltaValue, occurredAt, receivedAt, "", "")
|
||
|
||
bundle, err := s.loadEnabledBundle(ctx, sysOrigin)
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
|
||
matchedTasks := make([]model.TaskCenterTaskConfig, 0)
|
||
if bundle != nil {
|
||
for _, task := range bundle.Tasks {
|
||
if strings.EqualFold(task.ConditionType, eventType) {
|
||
matchedTasks = append(matchedTasks, task)
|
||
}
|
||
}
|
||
}
|
||
|
||
resultReason := ""
|
||
if bundle == nil {
|
||
resultReason = "task_center_disabled"
|
||
} else if len(matchedTasks) == 0 {
|
||
resultReason = "no_matched_task"
|
||
}
|
||
|
||
updated := 0
|
||
duplicate := false
|
||
if err := s.db.WithContext(ctx).Transaction(func(tx *gorm.DB) error {
|
||
inserted, err := insertTaskEventDedupTx(tx, sysOrigin, eventID, eventType, userID)
|
||
if err != nil {
|
||
return err
|
||
}
|
||
if !inserted {
|
||
duplicate = true
|
||
return nil
|
||
}
|
||
|
||
if err := createTaskEventArchiveOutboxTx(tx, sysOrigin, eventID, eventType, userID, occurredAt, rawJSON); err != nil {
|
||
return err
|
||
}
|
||
if resultReason != "" {
|
||
return updateTaskEventDedupTx(tx, sysOrigin, eventID, DedupStatusIgnored, resultReason)
|
||
}
|
||
|
||
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
|
||
}
|
||
updated++
|
||
}
|
||
if updated == 0 {
|
||
resultReason = "no_updated_task"
|
||
return updateTaskEventDedupTx(tx, sysOrigin, eventID, DedupStatusIgnored, resultReason)
|
||
}
|
||
return updateTaskEventDedupTx(tx, sysOrigin, eventID, DedupStatusSuccess, "")
|
||
}); err != nil {
|
||
return nil, err
|
||
}
|
||
|
||
if duplicate {
|
||
return &EventResponse{Processed: false, Updated: 0, Reason: "duplicate_event"}, nil
|
||
}
|
||
return &EventResponse{Processed: updated > 0, Updated: updated, Reason: resultReason}, nil
|
||
}
|
||
|
||
func insertTaskEventDedupTx(tx *gorm.DB, sysOrigin, eventID, eventType string, userID int64) (bool, error) {
|
||
id, err := utils.NextID()
|
||
if err != nil {
|
||
return false, err
|
||
}
|
||
now := time.Now()
|
||
row := model.TaskCenterEventDedup{
|
||
ID: id,
|
||
SysOrigin: sysOrigin,
|
||
EventID: eventID,
|
||
EventType: eventType,
|
||
UserID: userID,
|
||
Status: DedupStatusProcessing,
|
||
CreateTime: now,
|
||
UpdateTime: now,
|
||
}
|
||
result := tx.Clauses(clause.OnConflict{
|
||
Columns: []clause.Column{{Name: "sys_origin"}, {Name: "event_id"}},
|
||
DoNothing: true,
|
||
}).Create(&row)
|
||
if result.Error != nil {
|
||
return false, result.Error
|
||
}
|
||
return result.RowsAffected > 0, nil
|
||
}
|
||
|
||
func updateTaskEventDedupTx(tx *gorm.DB, sysOrigin, eventID, status, failureReason string) error {
|
||
return tx.Model(&model.TaskCenterEventDedup{}).
|
||
Where("sys_origin = ? AND event_id = ?", sysOrigin, eventID).
|
||
Updates(map[string]any{
|
||
"status": status,
|
||
"failure_reason": truncateTaskCenterFailure(failureReason),
|
||
"update_time": time.Now(),
|
||
}).Error
|
||
}
|
||
|
||
func createTaskEventArchiveOutboxTx(tx *gorm.DB, sysOrigin, eventID, eventType string, userID int64, occurredAt time.Time, rawJSON string) error {
|
||
id, err := utils.NextID()
|
||
if err != nil {
|
||
return err
|
||
}
|
||
now := time.Now()
|
||
row := model.TaskCenterEventArchiveOutbox{
|
||
ID: id,
|
||
SysOrigin: sysOrigin,
|
||
EventID: eventID,
|
||
EventType: eventType,
|
||
UserID: userID,
|
||
OccurredAt: occurredAt,
|
||
RawJSON: rawJSON,
|
||
Status: ArchiveStatusPending,
|
||
CreateTime: now,
|
||
UpdateTime: now,
|
||
}
|
||
return tx.Clauses(clause.OnConflict{
|
||
Columns: []clause.Column{{Name: "sys_origin"}, {Name: "event_id"}},
|
||
DoNothing: true,
|
||
}).Create(&row).Error
|
||
}
|
||
|
||
func buildArchiveRawJSON(req EventRequest, sysOrigin, eventID, eventType string, userID, deltaValue int64, occurredAt, receivedAt time.Time, processResult, processReason string) string {
|
||
row := map[string]any{
|
||
"schema_version": 1,
|
||
"sys_origin": sysOrigin,
|
||
"event_id": eventID,
|
||
"event_type": eventType,
|
||
"user_id": userID,
|
||
"delta_value": deltaValue,
|
||
"occurred_at": occurredAt.UTC().Format(time.RFC3339Nano),
|
||
"received_at": receivedAt.UTC().Format(time.RFC3339Nano),
|
||
"payload": req.Payload,
|
||
}
|
||
if processResult != "" {
|
||
row["process_result"] = processResult
|
||
}
|
||
if processReason != "" {
|
||
row["process_reason"] = processReason
|
||
}
|
||
bytes, err := json.Marshal(row)
|
||
if err != nil {
|
||
return "{}"
|
||
}
|
||
return string(bytes)
|
||
}
|
||
|
||
func truncateTaskCenterFailure(value string) string {
|
||
value = strings.TrimSpace(value)
|
||
if len(value) <= 512 {
|
||
return value
|
||
}
|
||
return value[:512]
|
||
}
|
||
|
||
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
|
||
}
|