Stop writing task center event rows
This commit is contained in:
ZuoZuo
parent
d39840a6ed
commit
df7446fb8c
@ -130,21 +130,4 @@ func registerTaskCenterRoutes(engine *gin.Engine, cfg config.Config, javaClient
|
||||
}
|
||||
writeOK(c, resp)
|
||||
})
|
||||
adminGroup.GET("/event/page", func(c *gin.Context) {
|
||||
resp, err := service.PageEvents(
|
||||
c.Request.Context(),
|
||||
c.Query("sysOrigin"),
|
||||
c.Query("taskCategory"),
|
||||
taskcenter.ParseInt64(c.Query("userId")),
|
||||
c.Query("taskCode"),
|
||||
c.Query("eventType"),
|
||||
int(taskcenter.ParseInt64(c.Query("cursor"))),
|
||||
int(taskcenter.ParseInt64(c.Query("limit"))),
|
||||
)
|
||||
if err != nil {
|
||||
writeError(c, err)
|
||||
return
|
||||
}
|
||||
writeOK(c, resp)
|
||||
})
|
||||
}
|
||||
|
||||
@ -233,45 +233,6 @@ func (s *Service) PageClaimRecords(ctx context.Context, sysOrigin string, taskCa
|
||||
return &PageResponse[ClaimRecordView]{Records: records, Total: total, Current: cursor, Size: limit}, nil
|
||||
}
|
||||
|
||||
// PageEvents 分页查询事件流水。
|
||||
func (s *Service) PageEvents(ctx context.Context, sysOrigin string, taskCategory string, userID int64, taskCode string, eventType string, cursor int, limit int) (*PageResponse[EventView], error) {
|
||||
sysOrigin = s.normalizeSysOrigin(sysOrigin)
|
||||
category := normalizeCategory(taskCategory)
|
||||
if err := validateCategory(category); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
cursor, limit = NormalizePage(cursor, limit)
|
||||
|
||||
query := s.db.WithContext(ctx).Model(&model.TaskCenterEvent{}).
|
||||
Where("sys_origin = ? AND task_category = ?", sysOrigin, category)
|
||||
if userID > 0 {
|
||||
query = query.Where("user_id = ?", userID)
|
||||
}
|
||||
if code := strings.ToUpper(strings.TrimSpace(taskCode)); code != "" {
|
||||
query = query.Where("task_code = ?", code)
|
||||
}
|
||||
if normalizedEventType := strings.ToUpper(strings.TrimSpace(eventType)); normalizedEventType != "" {
|
||||
query = query.Where("event_type = ?", normalizedEventType)
|
||||
}
|
||||
|
||||
var total int64
|
||||
if err := query.Count(&total).Error; err != nil {
|
||||
return nil, err
|
||||
}
|
||||
var rows []model.TaskCenterEvent
|
||||
if err := query.Order("create_time desc, id desc").
|
||||
Offset((cursor - 1) * limit).
|
||||
Limit(limit).
|
||||
Find(&rows).Error; err != nil {
|
||||
return nil, err
|
||||
}
|
||||
records := make([]EventView, 0, len(rows))
|
||||
for _, row := range rows {
|
||||
records = append(records, eventView(row))
|
||||
}
|
||||
return &PageResponse[EventView]{Records: records, Total: total, Current: cursor, Size: limit}, nil
|
||||
}
|
||||
|
||||
func (s *Service) normalizeTaskInputs(inputs []SaveTaskConfigInput, category string) ([]model.TaskCenterTaskConfig, error) {
|
||||
if len(inputs) == 0 {
|
||||
return nil, NewAppError(http.StatusBadRequest, "invalid_tasks", "tasks is required")
|
||||
@ -638,25 +599,6 @@ func claimRecordView(row model.TaskCenterClaimRecord) ClaimRecordView {
|
||||
}
|
||||
}
|
||||
|
||||
func eventView(row model.TaskCenterEvent) EventView {
|
||||
return EventView{
|
||||
ID: row.ID,
|
||||
SysOrigin: row.SysOrigin,
|
||||
EventID: row.EventID,
|
||||
EventType: row.EventType,
|
||||
UserID: row.UserID,
|
||||
TaskID: row.TaskID,
|
||||
TaskCode: row.TaskCode,
|
||||
TaskCategory: row.TaskCategory,
|
||||
CycleKey: row.CycleKey,
|
||||
DeltaValue: row.DeltaValue,
|
||||
CompletedValue: row.CompletedValue,
|
||||
OccurredAt: formatDateTime(row.OccurredAt),
|
||||
CreateTime: formatDateTime(row.CreateTime),
|
||||
UpdateTime: formatDateTime(row.UpdateTime),
|
||||
}
|
||||
}
|
||||
|
||||
func taskCenterSysOriginBySavedID(ctx context.Context, db taskCenterDB, configID int64, fallback string) string {
|
||||
if configID <= 0 {
|
||||
return fallback
|
||||
|
||||
@ -68,7 +68,6 @@ func (s *Service) ProcessEvent(ctx context.Context, req EventRequest) (*EventRes
|
||||
occurredAt := parseOccurredAt(req.OccurredAt, time.Now())
|
||||
receivedAt := time.Now()
|
||||
rawJSON := buildArchiveRawJSON(req, sysOrigin, eventID, eventType, userID, req.DeltaValue, occurredAt, receivedAt, "", "")
|
||||
payloadJSON := marshalPayloadJSON(req.Payload)
|
||||
|
||||
bundle, err := s.loadEnabledBundle(ctx, sysOrigin)
|
||||
if err != nil {
|
||||
@ -103,17 +102,6 @@ func (s *Service) ProcessEvent(ctx context.Context, req EventRequest) (*EventRes
|
||||
return nil
|
||||
}
|
||||
|
||||
var legacyEventCount int64
|
||||
if err := tx.Model(&model.TaskCenterEvent{}).
|
||||
Where("sys_origin = ? AND event_id = ?", sysOrigin, eventID).
|
||||
Count(&legacyEventCount).Error; err != nil {
|
||||
return err
|
||||
}
|
||||
if legacyEventCount > 0 {
|
||||
duplicate = true
|
||||
return updateTaskEventDedupTx(tx, sysOrigin, eventID, DedupStatusIgnored, "duplicate_event")
|
||||
}
|
||||
|
||||
if err := createTaskEventArchiveOutboxTx(tx, sysOrigin, eventID, eventType, userID, occurredAt, rawJSON); err != nil {
|
||||
return err
|
||||
}
|
||||
@ -141,31 +129,6 @@ func (s *Service) ProcessEvent(ctx context.Context, req EventRequest) (*EventRes
|
||||
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: payloadJSON,
|
||||
OccurredAt: occurredAt,
|
||||
CreateTime: time.Now(),
|
||||
UpdateTime: time.Now(),
|
||||
}
|
||||
if err := tx.Create(&eventRow).Error; err != nil {
|
||||
return err
|
||||
}
|
||||
updated++
|
||||
}
|
||||
if updated == 0 {
|
||||
@ -243,17 +206,6 @@ func createTaskEventArchiveOutboxTx(tx *gorm.DB, sysOrigin, eventID, eventType s
|
||||
}).Create(&row).Error
|
||||
}
|
||||
|
||||
func marshalPayloadJSON(payload map[string]any) string {
|
||||
if len(payload) == 0 {
|
||||
return ""
|
||||
}
|
||||
bytes, err := json.Marshal(payload)
|
||||
if err != nil {
|
||||
return ""
|
||||
}
|
||||
return string(bytes)
|
||||
}
|
||||
|
||||
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,
|
||||
|
||||
@ -127,6 +127,15 @@ func TestDailyTaskConfigEventListAndClaim(t *testing.T) {
|
||||
if outboxCount != 1 {
|
||||
t.Fatalf("archive outbox rows = %d, want 1", outboxCount)
|
||||
}
|
||||
var legacyEventCount int64
|
||||
if err := db.Model(&model.TaskCenterEvent{}).
|
||||
Where("sys_origin = ? AND event_id = ?", "LIKEI", "event-1").
|
||||
Count(&legacyEventCount).Error; err != nil {
|
||||
t.Fatalf("count legacy task event: %v", err)
|
||||
}
|
||||
if legacyEventCount != 0 {
|
||||
t.Fatalf("legacy task event rows = %d, want 0", legacyEventCount)
|
||||
}
|
||||
|
||||
tasks, err = service.ListTasks(ctx, user)
|
||||
if err != nil {
|
||||
|
||||
@ -234,23 +234,6 @@ type ClaimRecordView struct {
|
||||
UpdateTime string `json:"updateTime"`
|
||||
}
|
||||
|
||||
type EventView struct {
|
||||
ID int64 `json:"id,string"`
|
||||
SysOrigin string `json:"sysOrigin"`
|
||||
EventID string `json:"eventId"`
|
||||
EventType string `json:"eventType"`
|
||||
UserID int64 `json:"userId,string"`
|
||||
TaskID int64 `json:"taskId,string"`
|
||||
TaskCode string `json:"taskCode"`
|
||||
TaskCategory string `json:"taskCategory"`
|
||||
CycleKey string `json:"cycleKey"`
|
||||
DeltaValue int64 `json:"deltaValue"`
|
||||
CompletedValue int64 `json:"completedValue"`
|
||||
OccurredAt string `json:"occurredAt"`
|
||||
CreateTime string `json:"createTime"`
|
||||
UpdateTime string `json:"updateTime"`
|
||||
}
|
||||
|
||||
type flexibleInt64 int64
|
||||
|
||||
func (v *flexibleInt64) UnmarshalJSON(data []byte) error {
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user