package metrics

import (
	"testing"

	"github.com/prometheus/client_golang/prometheus"
	"github.com/prometheus/client_golang/prometheus/testutil"
	"github.com/stretchr/testify/require"
	"github.com/tidbcloud/dataflow-service/internal/db/entity"
)

func testDFJob() *entity.DFJob {
	job := &entity.DFJob{
		TenantID:  entity.ID(100),
		ClusterID: entity.ID(200),
	}
	job.ID = entity.ID(300)
	return job
}

func TestIncDMCreateStageFailure(t *testing.T) {
	reg := prometheus.NewRegistry()
	counter := prometheus.NewCounterVec(
		prometheus.CounterOpts{
			Namespace: DataflowNamespace,
			Subsystem: dmSubSystem,
			Name:      "create_stage_failed_total_test",
			Help:      "test counter",
		}, []string{TenantLabel, ClusterIDLabel, jobIDLabel, StageLabel, ReasonClassLabel})
	reg.MustRegister(counter)

	origCounter := DMCreateStageFailedTotal
	DMCreateStageFailedTotal = counter
	defer func() { DMCreateStageFailedTotal = origCounter }()

	job := testDFJob()

	IncDMCreateStageFailure(job, "dump_create", "create_task")
	require.InDelta(t, 1, testutil.ToFloat64(counter.WithLabelValues("100", "200", "300", "dump_create", "create_task")), 0.001)

	IncDMCreateStageFailure(job, "dump_start", "start_task")
	require.InDelta(t, 1, testutil.ToFloat64(counter.WithLabelValues("100", "200", "300", "dump_start", "start_task")), 0.001)

	IncDMCreateStageFailure(job, "dump_create", "create_task")
	require.InDelta(t, 2, testutil.ToFloat64(counter.WithLabelValues("100", "200", "300", "dump_create", "create_task")), 0.001)
}

func TestIncDMModifyFailure(t *testing.T) {
	reg := prometheus.NewRegistry()
	counter := prometheus.NewCounterVec(
		prometheus.CounterOpts{
			Namespace: DataflowNamespace,
			Subsystem: dmSubSystem,
			Name:      "modify_failed_total_test",
			Help:      "test counter",
		}, []string{TenantLabel, ClusterIDLabel, jobIDLabel, StepLabel, ReasonClassLabel})
	reg.MustRegister(counter)

	origCounter := DMModifyFailedTotal
	DMModifyFailedTotal = counter
	defer func() { DMModifyFailedTotal = origCounter }()

	job := testDFJob()

	steps := []struct {
		step        string
		reasonClass string
	}{
		{"load_job", "get_cr"},
		{"load_job", "get_sources"},
		{"load_spec", "get_modification"},
		{"load_spec", "get_spec"},
		{"scale_worker", "scale"},
		{"update_phase", "update"},
	}

	for _, s := range steps {
		IncDMModifyFailure(job, s.step, s.reasonClass)
		val := testutil.ToFloat64(counter.WithLabelValues("100", "200", "300", s.step, s.reasonClass))
		require.InDelta(t, 1, val, 0.001, "step=%s reason_class=%s", s.step, s.reasonClass)
	}
}

func TestDMCounterMetricsExist(t *testing.T) {
	require.NotNil(t, DMCreateStageFailedTotal)
	require.NotNil(t, DMModifyFailedTotal)
}

func TestNoMetricEmittedOnSuccess(t *testing.T) {
	reg := prometheus.NewRegistry()
	counter := prometheus.NewCounterVec(
		prometheus.CounterOpts{
			Namespace: DataflowNamespace,
			Subsystem: dmSubSystem,
			Name:      "modify_success_path_test",
			Help:      "test counter",
		}, []string{TenantLabel, ClusterIDLabel, jobIDLabel, StepLabel, ReasonClassLabel})
	reg.MustRegister(counter)

	origCounter := DMModifyFailedTotal
	DMModifyFailedTotal = counter
	defer func() { DMModifyFailedTotal = origCounter }()

	count, err := testutil.GatherAndCount(reg)
	require.NoError(t, err)
	require.Equal(t, 0, count, "no metrics should be emitted when no failures occur")
}