package metric

import (
	"context"
	"testing"
	"time"

	"github.com/prometheus/client_golang/prometheus"
	"github.com/prometheus/client_model/go"
	"github.com/stretchr/testify/require"
	"github.com/tidbcloud/pkgv2/bizerr"
	"github.com/tidbcloud/serverless-service/internal/domain/constant"
	"github.com/tidbcloud/serverless-service/internal/domain/entity"
	"github.com/tidbcloud/serverless-service/internal/domain/repo/mock_repo"
	"github.com/tidbcloud/serverless-service/internal/utils"
	"go.uber.org/mock/gomock"
)

func TestChangefeedMetricCollector_Collect_SkipNilAndExpiredFailedTime(t *testing.T) {
	ctrl := gomock.NewController(t)
	defer ctrl.Finish()

	repo := mock_repo.NewMockChangefeedRepo(ctrl)
	collector := NewChangefeedMetricCollector(repo)

	now := time.Now()
	repo.EXPECT().ListByFilter(gomock.Any(), gomock.Any()).DoAndReturn(
		func(_ context.Context, filter entity.ListChangefeedsFilter) ([]*entity.CDCChangefeed, bizerr.BizError) {
			if filter.State == constant.ChangfeedStateRunning {
				return []*entity.CDCChangefeed{}, nil
			}
			require.NotNil(t, filter.FailedTimeGTE)
			return []*entity.CDCChangefeed{
				{
					ID:         "cf-recent",
					ClusterID:  "c1",
					TenantID:   "t1",
					State:      constant.ChangefeedStateFailed,
					FailedTime: ptrTime(now.Add(-5 * time.Minute)),
				},
			}, nil
		},
	).Times(2)
	repo.EXPECT().GroupByState(gomock.Any()).Return([]*entity.ChangefeedStateCount{}, nil)

	ch := make(chan prometheus.Metric, 16)
	collector.Collect(ch)
	close(ch)

	var stateMetricCount int
	for metric := range ch {
		dtoMetric := &io_prometheus_client.Metric{}
		require.NoError(t, metric.Write(dtoMetric))
		labels := metricLabels(dtoMetric)
		if labels["changefeed_id"] != "" {
			stateMetricCount++
			require.Equal(t, "cf-recent", labels["changefeed_id"])
		}
	}
	require.Equal(t, 1, stateMetricCount)
}

func TestChangefeedMetricCollector_Collect_SkipUninitializedCheckpoint(t *testing.T) {
	ctrl := gomock.NewController(t)
	defer ctrl.Finish()

	repo := mock_repo.NewMockChangefeedRepo(ctrl)
	collector := NewChangefeedMetricCollector(repo)

	delayed := time.Now().Add(-2 * time.Hour)
	repo.EXPECT().ListByFilter(gomock.Any(), gomock.Any()).DoAndReturn(
		func(_ context.Context, filter entity.ListChangefeedsFilter) ([]*entity.CDCChangefeed, bizerr.BizError) {
			if filter.State != constant.ChangfeedStateRunning {
				return []*entity.CDCChangefeed{}, nil
			}
			return []*entity.CDCChangefeed{
				{
					// just created, TiCDC has not reported any checkpoint yet
					ID:            "cf-no-checkpoint",
					ClusterID:     "c1",
					TenantID:      "t1",
					State:         constant.ChangfeedStateRunning,
					CheckpointTSO: 0,
				},
				{
					ID:            "cf-delayed",
					ClusterID:     "c1",
					TenantID:      "t1",
					State:         constant.ChangfeedStateRunning,
					CheckpointTSO: utils.TimeToTSO(delayed),
				},
			}, nil
		},
	).Times(2)
	repo.EXPECT().GroupByState(gomock.Any()).Return([]*entity.ChangefeedStateCount{}, nil)

	ch := make(chan prometheus.Metric, 16)
	collector.Collect(ch)
	close(ch)

	delays := make(map[string]float64)
	for metric := range ch {
		dtoMetric := &io_prometheus_client.Metric{}
		require.NoError(t, metric.Write(dtoMetric))
		labels := metricLabels(dtoMetric)
		if labels["changefeed_id"] != "" {
			delays[labels["changefeed_id"]] = dtoMetric.GetGauge().GetValue()
		}
	}

	require.NotContains(t, delays, "cf-no-checkpoint")
	require.InDelta(t, 2*time.Hour.Seconds(), delays["cf-delayed"], 60)
}

func ptrTime(v time.Time) *time.Time {
	return &v
}

func metricLabels(m *io_prometheus_client.Metric) map[string]string {
	labels := make(map[string]string, len(m.GetLabel()))
	for _, label := range m.GetLabel() {
		labels[label.GetName()] = label.GetValue()
	}
	return labels
}