fix evalution

Author: zero3233-bd

backend/modules/evaluation/domain/entity/evaluation.go

@@ -88,7 +88,7 @@ func (p *FieldDisplayFormat) Scan(value interface{}) (err error) {
 	var result sql.NullInt64
 	err = result.Scan(value)
 	*p = FieldDisplayFormat(result.Int64)
-	return err
+	return
 }
 
 func (p *FieldDisplayFormat) Value() (driver.Value, error) {
@@ -130,7 +130,7 @@ func (p *FieldStatus) Scan(value interface{}) (err error) {
 	var result sql.NullInt64
 	err = result.Scan(value)
 	*p = FieldStatus(result.Int64)
-	return err
+	return
 }
 
 func (p *FieldStatus) Value() (driver.Value, error) {

backend/modules/evaluation/domain/entity/evaluation_set_schema.go

@@ -73,7 +73,7 @@ func (p *EvalTargetType) Scan(value interface{}) (err error) {
 	var result sql.NullInt64
 	err = result.Scan(value)
 	*p = EvalTargetType(result.Int64)
-	return err
+	return
 }
 
 func (p *EvalTargetType) Value() (driver.Value, error) {

backend/modules/evaluation/domain/entity/target.go

@@ -139,3 +139,142 @@ func TestTraceHubServiceImpl_SpanTriggerDispatchError(t *testing.T) {
 	require.Error(t, err)
 	require.ErrorContains(t, err, "invoke error")
 }
+
+func TestTraceHubServiceImpl_preDispatchHandlesUnstartedAndLimits(t *testing.T) {
+	ctrl := gomock.NewController(t)
+	t.Cleanup(ctrl.Finish)
+
+	mockRepo := repo_mocks.NewMockITaskRepo(ctrl)
+	stubProc := &stubProcessor{}
+
+	now := time.Now()
+	startAt := now.Add(-2 * time.Hour).UnixMilli()
+	endAt := now.Add(-time.Minute).UnixMilli()
+	workspaceID := int64(101)
+	taskID := int64(202)
+
+	sampl := &task.Sampler{
+		SampleRate:    floatPtr(1),
+		SampleSize:    int64Ptr(1),
+		IsCycle:       boolPtr(true),
+		CycleCount:    int64Ptr(1),
+		CycleInterval: int64Ptr(1),
+		CycleTimeUnit: ptr.Of(task.TimeUnitDay),
+	}
+	rule := &task.Rule{
+		EffectiveTime: &task.EffectiveTime{
+			StartAt: ptr.Of(startAt),
+			EndAt:   ptr.Of(endAt),
+		},
+		Sampler: sampl,
+	}
+
+	sub := &spanSubscriber{
+		taskID: taskID,
+		t: &task.Task{
+			ID:          ptr.Of(taskID),
+			WorkspaceID: ptr.Of(workspaceID),
+			TaskType:    task.TaskTypeAutoEval,
+			TaskStatus:  ptr.Of(task.TaskStatusUnstarted),
+			Rule:        rule,
+			BaseInfo:    &common.BaseInfo{},
+		},
+		processor: stubProc,
+		taskRepo:  mockRepo,
+		runType:   task.TaskRunTypeNewData,
+	}
+
+	taskRunConfig := &entity.TaskRun{
+		ID:          303,
+		TaskID:      taskID,
+		WorkspaceID: workspaceID,
+		TaskType:    task.TaskRunTypeNewData,
+		RunStatus:   task.TaskStatusRunning,
+		RunStartAt:  now.Add(-90 * time.Minute),
+		RunEndAt:    now.Add(-30 * time.Minute),
+	}
+
+	mockRepo.EXPECT().GetLatestNewDataTaskRun(gomock.Any(), gomock.AssignableToTypeOf(ptr.Of(int64(0))), taskID).Return(taskRunConfig, nil)
+	mockRepo.EXPECT().GetTaskCount(gomock.Any(), taskID).Return(int64(1), nil)
+	mockRepo.EXPECT().GetTaskRunCount(gomock.Any(), taskID, taskRunConfig.ID).Return(int64(1), nil)
+
+	impl := &TraceHubServiceImpl{taskRepo: mockRepo}
+	span := &loop_span.Span{
+		StartTime: now.UnixMilli(),
+		TraceID:   "trace",
+		SpanID:    "span",
+	}
+
+	err := impl.preDispatch(context.Background(), span, []*spanSubscriber{sub})
+	require.NoError(t, err)
+	require.Equal(t, 2, len(stubProc.createTaskRunReqs))
+	require.Equal(t, startAt, stubProc.createTaskRunReqs[0].RunStartAt)
+	require.True(t, stubProc.createTaskRunReqs[0].RunEndAt > startAt)
+	require.Equal(t, taskRunConfig.RunEndAt.UnixMilli(), stubProc.createTaskRunReqs[1].RunStartAt)
+	require.Equal(t, 1, stubProc.updateCallCount)
+	require.Equal(t, 4, stubProc.finishChangeInvoked)
+	require.Len(t, stubProc.finishChangeReqs, 4)
+	require.True(t, stubProc.finishChangeReqs[0].IsFinish)
+	require.True(t, stubProc.finishChangeReqs[1].IsFinish)
+	require.False(t, stubProc.finishChangeReqs[2].IsFinish)
+	require.False(t, stubProc.finishChangeReqs[3].IsFinish)
+}
+
+func TestTraceHubServiceImpl_preDispatchHandlesMissingTaskRunConfig(t *testing.T) {
+	ctrl := gomock.NewController(t)
+	t.Cleanup(ctrl.Finish)
+
+	mockRepo := repo_mocks.NewMockITaskRepo(ctrl)
+	stubProc := &stubProcessor{createTaskRunErr: errors.New("create run failed")}
+
+	now := time.Now()
+	startAt := now.Add(-10 * time.Minute).UnixMilli()
+	workspaceID := int64(303)
+	taskID := int64(404)
+
+	sampl := &task.Sampler{
+		IsCycle:       boolPtr(true),
+		CycleInterval: int64Ptr(2),
+		CycleTimeUnit: ptr.Of(task.TimeUnitWeek),
+	}
+	rule := &task.Rule{
+		EffectiveTime: &task.EffectiveTime{
+			StartAt: ptr.Of(startAt),
+			EndAt:   ptr.Of(now.Add(time.Hour).UnixMilli()),
+		},
+		Sampler: sampl,
+	}
+
+	sub := &spanSubscriber{
+		taskID: taskID,
+		t: &task.Task{
+			ID:          ptr.Of(taskID),
+			WorkspaceID: ptr.Of(workspaceID),
+			TaskType:    task.TaskTypeAutoEval,
+			TaskStatus:  ptr.Of(task.TaskStatusRunning),
+			Rule:        rule,
+			BaseInfo:    &common.BaseInfo{},
+		},
+		processor: stubProc,
+		taskRepo:  mockRepo,
+		runType:   task.TaskRunTypeNewData,
+	}
+
+	mockRepo.EXPECT().GetLatestNewDataTaskRun(gomock.Any(), gomock.AssignableToTypeOf(ptr.Of(int64(0))), taskID).Return(nil, nil)
+
+	impl := &TraceHubServiceImpl{taskRepo: mockRepo}
+	span := &loop_span.Span{
+		StartTime: now.UnixMilli(),
+		TraceID:   "trace",
+		SpanID:    "span",
+	}
+
+	err := impl.preDispatch(context.Background(), span, []*spanSubscriber{sub})
+	require.Error(t, err)
+	require.ErrorContains(t, err, "task run config not found")
+	require.Equal(t, 1, len(stubProc.createTaskRunReqs))
+	require.Equal(t, startAt, stubProc.createTaskRunReqs[0].RunStartAt)
+	expectedEnd := startAt + 2*7*24*time.Hour.Milliseconds()
+	require.Equal(t, expectedEnd, stubProc.createTaskRunReqs[0].RunEndAt)
+	require.Equal(t, 0, stubProc.finishChangeInvoked)
+}

backend/modules/observability/domain/task/service/taskexe/tracehub/span_trigger_test.go

@@ -26,6 +26,9 @@ type stubProcessor struct {
 	createTaskRunErr    error
 	finishChangeInvoked int
 	invokeCalled        bool
+	createTaskRunReqs   []taskexe.OnCreateTaskRunChangeReq
+	finishChangeReqs    []taskexe.OnFinishTaskChangeReq
+	updateCallCount     int
 }
 
 func (s *stubProcessor) ValidateConfig(context.Context, any) error {
@@ -42,15 +45,18 @@ func (s *stubProcessor) OnCreateTaskChange(context.Context, *entity.Observabilit
 }
 
 func (s *stubProcessor) OnUpdateTaskChange(context.Context, *entity.ObservabilityTask, string) error {
+	s.updateCallCount++
 	return s.updateErr
 }
 
-func (s *stubProcessor) OnFinishTaskChange(context.Context, taskexe.OnFinishTaskChangeReq) error {
+func (s *stubProcessor) OnFinishTaskChange(_ context.Context, req taskexe.OnFinishTaskChangeReq) error {
 	s.finishChangeInvoked++
+	s.finishChangeReqs = append(s.finishChangeReqs, req)
 	return s.finishErr
 }
 
-func (s *stubProcessor) OnCreateTaskRunChange(context.Context, taskexe.OnCreateTaskRunChangeReq) error {
+func (s *stubProcessor) OnCreateTaskRunChange(_ context.Context, req taskexe.OnCreateTaskRunChangeReq) error {
+	s.createTaskRunReqs = append(s.createTaskRunReqs, req)
 	return s.createTaskRunErr
 }