package impl

import (
	"context"
	"strconv"
	"testing"

	"github.com/prometheus/client_golang/prometheus"
	dto "github.com/prometheus/client_model/go"
	"github.com/stretchr/testify/assert"
	"github.com/stretchr/testify/mock"
	"github.com/stretchr/testify/suite"
	app_metric "github.com/tidbcloud/dataflow-service/app/metrics/changefeed/metric"
	"github.com/tidbcloud/dataflow-service/app/models/domain"
	ticdc_mock "github.com/tidbcloud/dataflow-service/app/models/ticdc/mock"
	"github.com/tidbcloud/dataflow-service/app/repo/entity"
	repo_mock "github.com/tidbcloud/dataflow-service/app/repo/mock"
	internal_metrics "github.com/tidbcloud/dataflow-service/internal/metrics"
	errhandler "github.com/tidbcloud/dataflow-service/pkg/error-handler"
	"github.com/tidbcloud/pkgv2/bizerr"
	"go.uber.org/zap"
	"go.uber.org/zap/zaptest/observer"
)

type bizChangefeedMetricsWorkerTestSuite struct {
	suite.Suite

	worker *BizChangefeedMetricsWorkerImpl

	mockBizChangefeedRepo     *repo_mock.BizChangefeedRepo
	mockKernelChangefeedModel *ticdc_mock.ChangefeedModel
}

func (s *bizChangefeedMetricsWorkerTestSuite) SetupSuite() {
	s.mockBizChangefeedRepo = repo_mock.NewBizChangefeedRepo(s.T())
	s.mockKernelChangefeedModel = ticdc_mock.NewChangefeedModel(s.T())
	s.worker = &BizChangefeedMetricsWorkerImpl{
		bizChangefeedRepo:     s.mockBizChangefeedRepo,
		kernelChangefeedModel: s.mockKernelChangefeedModel,
	}
}

func (s *bizChangefeedMetricsWorkerTestSuite) TestUpdateChangefeedMetrics() {
	s.Run("update and clean metrics error", func() {
		mockErr := bizerr.New(bizerr.ErrCodeDB, "network error")
		listCall := s.mockBizChangefeedRepo.On("List", mock.Anything).Return(nil, mockErr)
		defer listCall.Unset()
		listDeleted := s.mockBizChangefeedRepo.
			On("ListWithSoftDelete", mock.Anything, mock.Anything).
			Return(nil, mockErr)
		defer listDeleted.Unset()

		err := s.worker.UpdateChangefeedMetrics(context.Background())
		s.Error(err)
		s.Contains(err.Error(), "update changefeed metrics error")
		s.Contains(err.Error(), "clean unused changefeed metrics error")
	})

	s.Run("update metrics successfully but clean error", func() {
		listCall := s.mockBizChangefeedRepo.On("List", mock.Anything).Return(nil, nil)
		defer listCall.Unset()
		mockErr := bizerr.New(bizerr.ErrCodeDB, "network error")
		listDeleted := s.mockBizChangefeedRepo.
			On("ListWithSoftDelete", mock.Anything, mock.Anything).
			Return(nil, mockErr)
		defer listDeleted.Unset()

		err := s.worker.UpdateChangefeedMetrics(context.Background())
		s.Error(err)
		s.NotContains(err.Error(), "update changefeed metrics error")
		s.Contains(err.Error(), "clean unused changefeed metrics error")
	})

	s.Run("clean metrics successfully but update error", func() {
		listDeleted := s.mockBizChangefeedRepo.
			On("ListWithSoftDelete", mock.Anything, mock.Anything).
			Return(nil, nil)
		defer listDeleted.Unset()
		mockErr := bizerr.New(bizerr.ErrCodeDB, "network error")
		listCall := s.mockBizChangefeedRepo.On("List", mock.Anything).Return(nil, mockErr)
		defer listCall.Unset()

		err := s.worker.UpdateChangefeedMetrics(context.Background())
		s.Error(err)
		s.Contains(err.Error(), "update changefeed metrics error")
		s.NotContains(err.Error(), "clean unused changefeed metrics error")
	})

	s.Run("update and clean metrics successfully", func() {
		listCall := s.mockBizChangefeedRepo.On("List", mock.Anything).Return(nil, nil)
		defer listCall.Unset()
		listDeleted := s.mockBizChangefeedRepo.
			On("ListWithSoftDelete", mock.Anything, mock.Anything).
			Return(nil, nil)
		defer listDeleted.Unset()

		err := s.worker.UpdateChangefeedMetrics(context.Background())
		s.NoError(err)
	})
}

func (s *bizChangefeedMetricsWorkerTestSuite) TestUpdateLiveChangefeedMetrics() {
	mockClusterID := entity.ID(191)
	logger := zap.NewNop()
	defer func() {
		app_metric.ChangefeedStatus.DeletePartialMatch(prometheus.Labels{
			app_metric.ClusterIDLabel: strconv.FormatUint(mockClusterID.Uint64(), 10),
		})
	}()
	s.Run("list biz_changefeed ent error", func() {
		mockErr := bizerr.New(bizerr.ErrCodeDB, "network error")
		call := s.mockBizChangefeedRepo.On("List", mock.Anything).Return(nil, mockErr)
		defer call.Unset()
		err := s.worker.updateLiveChangefeedMetrics(context.Background(), logger)
		s.Error(err)
		s.Contains(err.Error(), "list all live biz_changefeeds")
	})

	s.Run("no live biz_changefeed", func() {
		call := s.mockBizChangefeedRepo.On("List", mock.Anything).Return(nil, nil)
		defer call.Unset()
		err := s.worker.updateLiveChangefeedMetrics(context.Background(), logger)
		s.NoError(err)
	})

	mockEnts := []entity.BizChangefeed{
		{
			ID:              1111,
			ClusterID:       mockClusterID,
			OperationStatus: entity.BizChangefeedOperationStatusNormal,
		},
		{
			ID:              1112,
			ClusterID:       mockClusterID,
			OperationStatus: entity.BizChangefeedOperationStatusCreating,
		},
	}
	listEntCall := s.mockBizChangefeedRepo.On("List", mock.Anything).Return(mockEnts, nil)
	defer listEntCall.Unset()
	s.Run("list kernel changefeed error", func() {
		mockErr := bizerr.New(bizerr.ErrCodeUnknown, "network error")
		call := s.mockKernelChangefeedModel.
			On("ListChangefeeds", mock.Anything, mock.Anything).
			Return(nil, mockErr)
		defer call.Unset()
		err := s.worker.updateLiveChangefeedMetrics(context.Background(), logger)
		s.Error(err)
		s.Contains(err.Error(), "list all live kernel changefeeds")
	})

	mockKernelChangefeeds := []domain.ListChangefeedItem{
		{
			ChangefeedEntity: entity.TiCDCChangefeed{
				BusinessID: 1111,
				KernelMetrics: entity.TiCDCChangefeedKernelMetrics{
					FeedState: entity.ChangefeedKernelStatusNormal,
				},
			},
		},
	}
	mockResp := &domain.ListChangefeedsResp{
		Items: mockKernelChangefeeds,
	}
	listKernelCall := s.mockKernelChangefeedModel.
		On("ListChangefeeds", mock.Anything, mock.Anything).
		Return(mockResp, nil)
	defer listKernelCall.Unset()
	s.Run("update changefeed metrics successfully", func() {
		err := s.worker.updateLiveChangefeedMetrics(context.Background(), logger)
		s.NoError(err)
		matcher := mock.MatchedBy(func(req *domain.ListChangefeedsReq) bool {
			return req.Filter.Deleted == false && *req.Filter.BizType == domain.TiCDCClusterBizTypeChangefeed
		})
		s.mockKernelChangefeedModel.AssertCalled(s.T(), "ListChangefeeds", mock.Anything, matcher)
	})
}

func (s *bizChangefeedMetricsWorkerTestSuite) TestCleanUnusedChangefeedMetrics() {
	mockClusterID := entity.ID(192)
	mockBizChangefeedID := entity.ID(1111)
	logger := zap.NewNop()
	app_metric.ChangefeedStatus.
		WithLabelValues(
			strconv.FormatUint(mockClusterID.Uint64(), 10),
			strconv.FormatUint(mockBizChangefeedID.Uint64(), 10)).
		Set(1)
	defer func() {
		app_metric.ChangefeedStatus.DeletePartialMatch(prometheus.Labels{
			app_metric.ClusterIDLabel: strconv.FormatUint(mockClusterID.Uint64(), 10),
		})
	}()

	s.Run("list deleted biz_changefeed error", func() {
		mockErr := bizerr.New(bizerr.ErrCodeDB, "network error")
		call := s.mockBizChangefeedRepo.
			On("ListWithSoftDelete", mock.Anything, mock.Anything).
			Return(nil, mockErr)
		defer call.Unset()
		err := s.worker.cleanUnusedChangefeedMetrics(context.Background(), logger)
		s.Error(err)
		s.Contains(err.Error(), "list deleted biz_changefeeds")
	})

	deletedBizEnts := []entity.BizChangefeed{
		{
			ID:        mockBizChangefeedID,
			ClusterID: mockClusterID,
		},
		{
			ID:        mockBizChangefeedID + 1,
			ClusterID: mockClusterID,
		},
	}
	call := s.mockBizChangefeedRepo.
		On("ListWithSoftDelete", mock.Anything, mock.Anything).
		Return(deletedBizEnts, nil)
	defer call.Unset()
	s.Run("clean metrics successfully", func() {
		err := s.worker.cleanUnusedChangefeedMetrics(context.Background(), logger)
		s.NoError(err)
		metric := new(dto.Metric)
		err = app_metric.ChangefeedStatus.
			WithLabelValues(
				strconv.FormatUint(mockClusterID.Uint64(), 10),
				strconv.FormatUint(mockBizChangefeedID.Uint64(), 10)).
			Write(metric)
		s.NoError(err)
		s.Equal(float64(0), metric.GetGauge().GetValue())
	})
}

func (s *bizChangefeedMetricsWorkerTestSuite) TestUpdateMetricsForBizChangefeed() {
	mockClusterID := entity.ID(193)
	testcases := []struct {
		bizChangefeedEnt    *entity.BizChangefeed
		kernelChangefeedEnt *entity.TiCDCChangefeed
	}{
		{
			bizChangefeedEnt: &entity.BizChangefeed{
				ID:              1111,
				ClusterID:       mockClusterID,
				OperationStatus: entity.BizChangefeedOperationStatusDeleting,
			},
			kernelChangefeedEnt: nil,
		},
		{
			bizChangefeedEnt: &entity.BizChangefeed{
				ID:              1112,
				ClusterID:       mockClusterID,
				OperationStatus: entity.BizChangefeedOperationStatusNormal,
			},
			kernelChangefeedEnt: &entity.TiCDCChangefeed{
				KernelMetrics: entity.TiCDCChangefeedKernelMetrics{
					FeedState: entity.ChangefeedKernelStatusWarning,
				},
			},
		},
		{
			bizChangefeedEnt: &entity.BizChangefeed{
				ID:              1113,
				ClusterID:       mockClusterID,
				OperationStatus: entity.BizChangefeedOperationStatusFailed,
				ErrMsg:          "create failed",
			},
			kernelChangefeedEnt: nil,
		},
	}

	logger := zap.NewNop()
	defer func() {
		app_metric.ChangefeedStatus.DeletePartialMatch(prometheus.Labels{
			app_metric.ClusterIDLabel: strconv.FormatUint(mockClusterID.Uint64(), 10),
		})
		internal_metrics.ChangefeedCreateResultCounter.DeletePartialMatch(prometheus.Labels{
			internal_metrics.ClusterIDLabel: strconv.FormatUint(mockClusterID.Uint64(), 10),
		})
	}()

	for _, tc := range testcases {
		s.worker.updateMetricsForBizChangefeed(logger, tc.bizChangefeedEnt, tc.kernelChangefeedEnt)
		statusMetric := new(dto.Metric)
		err := app_metric.ChangefeedStatus.
			WithLabelValues(
				strconv.FormatUint(tc.bizChangefeedEnt.ClusterID.Uint64(), 10),
				strconv.FormatUint(tc.bizChangefeedEnt.ID.Uint64(), 10)).
			Write(statusMetric)
		s.NoError(err)
		s.NotEqual(float64(0), statusMetric.GetGauge().GetValue())
		if tc.bizChangefeedEnt.OperationStatus == entity.BizChangefeedOperationStatusFailed {
			createResultMetric := new(dto.Metric)
			err := internal_metrics.ChangefeedCreateResultCounter.
				WithLabelValues(
					strconv.FormatUint(tc.bizChangefeedEnt.ClusterID.Uint64(), 10),
					strconv.FormatUint(tc.bizChangefeedEnt.ClusterID.Uint64(), 10),
					genChangefeedIDStrInMetric(tc.bizChangefeedEnt.ID.Uint64()),
					internal_metrics.Dedicate,
					internal_metrics.ChangefeedCreateResultFailed,
					tc.bizChangefeedEnt.ErrMsg,
					errhandler.ChangefeedErrType(tc.bizChangefeedEnt.ErrMsg)).
				Write(createResultMetric)
			s.NoError(err)
			s.NotEqual(float64(0), createResultMetric.GetCounter().GetValue())
		}
	}
}

func TestBizChangefeedMetricsWorkerTestSuite(t *testing.T) {
	suite.Run(t, new(bizChangefeedMetricsWorkerTestSuite))
}

func TestConvertChangefeedStatus(t *testing.T) {
	testcases := []struct {
		bizChangefeedEnt    *entity.BizChangefeed
		kernelChangefeedEnt *entity.TiCDCChangefeed
		expectedStatus      entity.CDCStatus
	}{
		{
			bizChangefeedEnt:    &entity.BizChangefeed{},
			kernelChangefeedEnt: nil,
			expectedStatus:      entity.CDCStatusNormal,
		},
		{
			bizChangefeedEnt: &entity.BizChangefeed{
				OperationStatus: entity.BizChangefeedOperationStatusCreating,
			},
			kernelChangefeedEnt: nil,
			expectedStatus:      entity.CDCStatusCreating,
		},
		{
			bizChangefeedEnt: &entity.BizChangefeed{
				OperationStatus: entity.BizChangefeedOperationStatusCreating,
			},
			kernelChangefeedEnt: &entity.TiCDCChangefeed{},
			expectedStatus:      entity.CDCStatusCreating,
		},
		{
			bizChangefeedEnt: &entity.BizChangefeed{
				OperationStatus: entity.BizChangefeedOperationStatusCreating,
			},
			kernelChangefeedEnt: &entity.TiCDCChangefeed{
				KernelMetrics: entity.TiCDCChangefeedKernelMetrics{
					FeedState: entity.ChangefeedKernelStatusNormal,
				},
			},
			expectedStatus: entity.CDCStatusCreating,
		},
		{
			bizChangefeedEnt: &entity.BizChangefeed{
				OperationStatus: entity.BizChangefeedOperationStatusFailed,
			},
			kernelChangefeedEnt: nil,
			expectedStatus:      entity.CDCStatusFailed,
		},
		{
			bizChangefeedEnt: &entity.BizChangefeed{
				OperationStatus: entity.BizChangefeedOperationStatusNormal,
			},
			kernelChangefeedEnt: &entity.TiCDCChangefeed{
				KernelMetrics: entity.TiCDCChangefeedKernelMetrics{
					FeedState: entity.ChangefeedKernelStatusNormal,
				},
			},
			expectedStatus: entity.CDCStatusNormal,
		},
		{
			bizChangefeedEnt: &entity.BizChangefeed{
				OperationStatus: entity.BizChangefeedOperationStatusScaling,
			},
			kernelChangefeedEnt: &entity.TiCDCChangefeed{
				KernelMetrics: entity.TiCDCChangefeedKernelMetrics{
					FeedState: entity.ChangefeedKernelStatusNormal,
				},
			},
			expectedStatus: entity.CDCStatusScaling,
		},
		{
			bizChangefeedEnt: &entity.BizChangefeed{
				OperationStatus: entity.BizChangefeedOperationStatusPaused,
			},
			kernelChangefeedEnt: &entity.TiCDCChangefeed{
				KernelMetrics: entity.TiCDCChangefeedKernelMetrics{
					FeedState: entity.ChangefeedKernelStatusStopped,
				},
			},
			expectedStatus: entity.CDCStatusStopped,
		},
		{
			bizChangefeedEnt: &entity.BizChangefeed{
				OperationStatus: entity.BizChangefeedOperationStatusDeleting,
			},
			kernelChangefeedEnt: &entity.TiCDCChangefeed{
				KernelMetrics: entity.TiCDCChangefeedKernelMetrics{
					FeedState: entity.ChangefeedKernelStatusNormal,
				},
			},
			expectedStatus: entity.CDCStatusDeleting,
		},
		{
			bizChangefeedEnt: &entity.BizChangefeed{
				OperationStatus: entity.BizChangefeedOperationStatusDeleted,
			},
			kernelChangefeedEnt: &entity.TiCDCChangefeed{
				KernelMetrics: entity.TiCDCChangefeedKernelMetrics{
					FeedState: entity.ChangefeedKernelStatusNormal,
				},
			},
			expectedStatus: entity.CDCStatusDeleted,
		},
		{
			bizChangefeedEnt: &entity.BizChangefeed{
				OperationStatus: entity.BizChangefeedOperationStatusNormal,
			},
			kernelChangefeedEnt: &entity.TiCDCChangefeed{
				KernelMetrics: entity.TiCDCChangefeedKernelMetrics{
					FeedState: entity.ChangefeedKernelStatusWarning,
				},
			},
			expectedStatus: entity.CDCStatusError,
		},
		{
			bizChangefeedEnt: &entity.BizChangefeed{
				OperationStatus: entity.BizChangefeedOperationStatusNormal,
			},
			kernelChangefeedEnt: &entity.TiCDCChangefeed{
				KernelMetrics: entity.TiCDCChangefeedKernelMetrics{
					FeedState: entity.ChangefeedKernelStatusError,
				},
			},
			expectedStatus: entity.CDCStatusError,
		},
		{
			bizChangefeedEnt: &entity.BizChangefeed{
				OperationStatus: entity.BizChangefeedOperationStatusNormal,
			},
			kernelChangefeedEnt: &entity.TiCDCChangefeed{
				KernelMetrics: entity.TiCDCChangefeedKernelMetrics{
					FeedState: entity.ChangefeedKernelStatusFailed,
				},
			},
			expectedStatus: entity.CDCStatusFailed,
		},
		{
			bizChangefeedEnt: &entity.BizChangefeed{
				OperationStatus: entity.BizChangefeedOperationStatusNormal,
			},
			kernelChangefeedEnt: &entity.TiCDCChangefeed{
				KernelMetrics: entity.TiCDCChangefeedKernelMetrics{
					FeedState: entity.ChangefeedKernelStatusStopped,
				},
			},
			expectedStatus: entity.CDCStatusStopped,
		},
	}

	for _, tc := range testcases {
		cdcStatus := convertChangefeedStatus(tc.bizChangefeedEnt, tc.kernelChangefeedEnt)
		assert.Equal(t, tc.expectedStatus, cdcStatus)
	}
}

// TestUpdateLiveChangefeedMetricsPagination verifies that when ListChangefeeds returns a full
// page the worker continues to fetch subsequent pages, collecting all kernel changefeeds.
func TestUpdateLiveChangefeedMetricsPagination(t *testing.T) {
	mockBizChangefeedRepo := repo_mock.NewBizChangefeedRepo(t)
	mockKernelChangefeedModel := ticdc_mock.NewChangefeedModel(t)
	worker := &BizChangefeedMetricsWorkerImpl{
		bizChangefeedRepo:     mockBizChangefeedRepo,
		kernelChangefeedModel: mockKernelChangefeedModel,
	}

	mockClusterID := entity.ID(200)
	logger := zap.NewNop()
	defer func() {
		app_metric.ChangefeedStatus.DeletePartialMatch(prometheus.Labels{
			app_metric.ClusterIDLabel: strconv.FormatUint(mockClusterID.Uint64(), 10),
		})
	}()

	// bizID only appears in page 2 — its metric being set proves pagination worked.
	bizID := entity.ID(99999)
	mockBizChangefeedRepo.On("List", mock.Anything).Return([]entity.BizChangefeed{
		{ID: bizID, ClusterID: mockClusterID, OperationStatus: entity.BizChangefeedOperationStatusNormal},
	}, nil)

	// Full first page (none of the items has bizID).
	page1Items := make([]domain.ListChangefeedItem, listAllKernelChangefeedsPageSize)
	for i := 0; i < listAllKernelChangefeedsPageSize; i++ {
		page1Items[i] = domain.ListChangefeedItem{
			ChangefeedEntity: entity.TiCDCChangefeed{
				BusinessID: entity.ID(100000 + i),
				KernelMetrics: entity.TiCDCChangefeedKernelMetrics{
					FeedState: entity.ChangefeedKernelStatusNormal,
				},
			},
		}
	}

	// Partial second page contains bizID.
	page2Items := []domain.ListChangefeedItem{
		{
			ChangefeedEntity: entity.TiCDCChangefeed{
				BusinessID: bizID,
				KernelMetrics: entity.TiCDCChangefeedKernelMetrics{
					FeedState: entity.ChangefeedKernelStatusNormal,
				},
			},
		},
	}

	mockKernelChangefeedModel.On("ListChangefeeds", mock.Anything,
		mock.MatchedBy(func(req *domain.ListChangefeedsReq) bool {
			return req.Pager != nil && req.Pager.Page == 0
		})).Return(&domain.ListChangefeedsResp{Items: page1Items}, nil).Once()
	mockKernelChangefeedModel.On("ListChangefeeds", mock.Anything,
		mock.MatchedBy(func(req *domain.ListChangefeedsReq) bool {
			return req.Pager != nil && req.Pager.Page == 1
		})).Return(&domain.ListChangefeedsResp{Items: page2Items}, nil).Once()

	err := worker.updateLiveChangefeedMetrics(context.Background(), logger)
	assert.NoError(t, err)

	// The biz changefeed from page 2 should have its metric set.
	statusMetric := new(dto.Metric)
	err = app_metric.ChangefeedStatus.WithLabelValues(
		strconv.FormatUint(mockClusterID.Uint64(), 10),
		strconv.FormatUint(bizID.Uint64(), 10)).
		Write(statusMetric)
	assert.NoError(t, err)
	assert.NotEqual(t, float64(0), statusMetric.GetGauge().GetValue(),
		"biz changefeed from page 2 should have its metric set, proving all pages were fetched")
	mockKernelChangefeedModel.AssertExpectations(t)
}

// TestUpdateMetricsWarnGuard verifies that the "kernel changefeed not found" WARN is only
// emitted for statuses where the kernel changefeed is expected to exist (normal/scaling/paused),
// and NOT for transient statuses (creating/deleting/deleted/failed).
func TestUpdateMetricsWarnGuard(t *testing.T) {
	testcases := []struct {
		name            string
		operationStatus entity.BizChangefeedOperationStatus
		expectWarn      bool
	}{
		{"normal+nil kernel should warn", entity.BizChangefeedOperationStatusNormal, true},
		{"scaling+nil kernel should warn", entity.BizChangefeedOperationStatusScaling, true},
		{"paused+nil kernel should warn", entity.BizChangefeedOperationStatusPaused, true},
		{"creating+nil kernel should not warn", entity.BizChangefeedOperationStatusCreating, false},
		{"deleting+nil kernel should not warn", entity.BizChangefeedOperationStatusDeleting, false},
		{"deleted+nil kernel should not warn", entity.BizChangefeedOperationStatusDeleted, false},
		{"failed+nil kernel should not warn", entity.BizChangefeedOperationStatusFailed, false},
	}

	for _, tc := range testcases {
		t.Run(tc.name, func(t *testing.T) {
			core, logs := observer.New(zap.WarnLevel)
			logger := zap.New(core)
			worker := &BizChangefeedMetricsWorkerImpl{}
			bizEnt := &entity.BizChangefeed{
				ID:              entity.ID(1),
				ClusterID:       entity.ID(1),
				OperationStatus: tc.operationStatus,
			}
			defer app_metric.ChangefeedStatus.DeletePartialMatch(prometheus.Labels{
				app_metric.ClusterIDLabel: strconv.FormatUint(bizEnt.ClusterID.Uint64(), 10),
			})

			worker.updateMetricsForBizChangefeed(logger, bizEnt, nil)

			warnCount := logs.FilterMessage("kernel changefeed not found").Len()
			if tc.expectWarn {
				assert.Greater(t, warnCount, 0,
					"expected WARN log for status %q with nil kernel", tc.operationStatus)
			} else {
				assert.Equal(t, 0, warnCount,
					"unexpected WARN log for status %q with nil kernel", tc.operationStatus)
			}
		})
	}
}