package impl

import (
	"context"
	"fmt"
	"strconv"
	"time"

	"github.com/prometheus/client_golang/prometheus"
	app_metric "github.com/tidbcloud/dataflow-service-ng/app/metrics/changefeed/metric"
	"github.com/tidbcloud/dataflow-service-ng/app/models/biz_changefeed"
	"github.com/tidbcloud/dataflow-service-ng/app/models/domain"
	"github.com/tidbcloud/dataflow-service-ng/app/models/ticdc"
	"github.com/tidbcloud/dataflow-service-ng/app/repo"
	"github.com/tidbcloud/dataflow-service-ng/app/repo/entity"
	internal_metrics "github.com/tidbcloud/dataflow-service-ng/internal/metrics"
	errwrapper "github.com/tidbcloud/dataflow-service-ng/pkg/error-wrapper"
	"github.com/tidbcloud/dataflow-service-ng/pkg/log"
	"go.uber.org/multierr"
	"go.uber.org/zap"
)

const (
	changefeedCreateResultFailed     = "failed"
	softDeletedRecordsRetrieveTime   = 15 * time.Minute
	listAllKernelChangefeedsPageSize = 1000
)

type BizChangefeedMetricsWorkerImpl struct {
	kernelChangefeedModel ticdc.ChangefeedModel
	bizChangefeedRepo     repo.BizChangefeedRepo
}

func NewBizChangefeedMetricsWorker(
	kernelChangefeedModel ticdc.ChangefeedModel,
	bizChangefeedRepo repo.BizChangefeedRepo,
) biz_changefeed.BizChangefeedMetricsWorker {
	return &BizChangefeedMetricsWorkerImpl{
		kernelChangefeedModel: kernelChangefeedModel,
		bizChangefeedRepo:     bizChangefeedRepo,
	}
}

func (m *BizChangefeedMetricsWorkerImpl) UpdateChangefeedMetrics(ctx context.Context) error {
	ctx = log.WithTraceID(ctx)
	logger := log.GetLogger(ctx).Named("UpdateChangefeedMetrics")
	var multiErr error
	if err := m.updateLiveChangefeedMetrics(ctx, logger); err != nil {
		multiErr = multierr.Append(multiErr, fmt.Errorf("update changefeed metrics error: %w", err))
	}
	if err := m.cleanUnusedChangefeedMetrics(ctx, logger); err != nil {
		multiErr = multierr.Append(multiErr, fmt.Errorf("clean unused changefeed metrics error: %w", err))
	}
	return multiErr
}

func (m *BizChangefeedMetricsWorkerImpl) updateLiveChangefeedMetrics(ctx context.Context, logger *zap.Logger) error {
	allLiveBizChangefeeds, bizErr := m.bizChangefeedRepo.List(ctx)
	if bizErr != nil {
		logger.Error("list all live biz_changefeeds error", zap.Error(bizErr))
		return bizErr.Chain("list all live biz_changefeeds")
	}
	if len(allLiveBizChangefeeds) == 0 {
		logger.Info("no live biz_changefeed found")
		return nil
	}

	bizID2KernelChangefees := make(map[uint64]*entity.TiCDCChangefeed)
	for page := int64(0); ; page++ {
		resp, bizErr := m.kernelChangefeedModel.ListChangefeeds(ctx, &domain.ListChangefeedsReq{
			Filter: &domain.ListChangefeedFilter{
				Deleted: false,
			},
			Pager: &domain.Pager{Page: page, PageSize: listAllKernelChangefeedsPageSize},
		})
		if bizErr != nil {
			logger.Error("list all live kernel changefeeds error", zap.Error(bizErr))
			return bizErr.Chain("list all live kernel changefeeds")
		}
		for i := range resp.Items {
			bizID2KernelChangefees[resp.Items[i].ChangefeedEntity.BizChangefeedID.Uint64()] = &resp.Items[i].ChangefeedEntity
		}
		if len(resp.Items) < listAllKernelChangefeedsPageSize {
			break
		}
	}

	for i := range allLiveBizChangefeeds {
		m.updateMetricsForBizChangefeed(
			logger, &allLiveBizChangefeeds[i], bizID2KernelChangefees[allLiveBizChangefeeds[i].ID.Uint64()])
	}
	return nil
}

func (m *BizChangefeedMetricsWorkerImpl) cleanUnusedChangefeedMetrics(ctx context.Context, logger *zap.Logger) error {
	deletedBizChangefeeds, bizErr := m.bizChangefeedRepo.ListWithSoftDelete(
		ctx, func(qs entity.BizChangefeedQuerySet) entity.BizChangefeedQuerySet {
			return qs.OperationStatusEq(entity.BizChangefeedOperationStatusDeleted).
				UpdatedAtGte(time.Now().Add(-softDeletedRecordsRetrieveTime))
		})
	if bizErr != nil {
		logger.Error("list deleted biz_changefeeds error", zap.Error(bizErr))
		return bizErr.Chain("list deleted biz_changefeeds")
	}

	for i := range deletedBizChangefeeds {
		deletedCount := app_metric.ChangefeedStatus.DeletePartialMatch(prometheus.Labels{
			app_metric.ChangefeedIDLabel: strconv.FormatUint(deletedBizChangefeeds[i].ID.Uint64(), 10),
		})
		if deletedCount > 0 {
			logger.Info("delete changefeed status metric for biz changefeed",
				zap.Uint64("biz_changefeed_id", deletedBizChangefeeds[i].ID.Uint64()),
				zap.Int("deleted_count", deletedCount))
		}
	}
	return nil
}

func (m *BizChangefeedMetricsWorkerImpl) updateMetricsForBizChangefeed(
	logger *zap.Logger, bizChangefeedEnt *entity.BizChangefeed, kernelChangefeedEnt *entity.TiCDCChangefeed,
) {
	kernelExpectedStatuses := map[entity.BizChangefeedOperationStatus]struct{}{
		entity.BizChangefeedOperationStatusNormal:  {},
		entity.BizChangefeedOperationStatusScaling: {},
		entity.BizChangefeedOperationStatusPaused:  {},
	}
	if kernelChangefeedEnt == nil {
		if _, expected := kernelExpectedStatuses[bizChangefeedEnt.OperationStatus]; expected {
			logger.Warn("kernel changefeed not found",
				zap.Uint64("biz_changefeed_id", bizChangefeedEnt.ID.Uint64()),
				zap.String("operation_status", string(bizChangefeedEnt.OperationStatus)))
		}
	}

	app_metric.ChangefeedStatus.WithLabelValues(
		strconv.FormatUint(bizChangefeedEnt.ClusterID.Uint64(), 10),
		strconv.FormatUint(bizChangefeedEnt.ID.Uint64(), 10)).
		Set(float64(app_metric.ChangefeedStatusIntMap[convertChangefeedStatus(bizChangefeedEnt, kernelChangefeedEnt)]))

	if bizChangefeedEnt.OperationStatus == entity.BizChangefeedOperationStatusFailed {
		internal_metrics.ChangefeedCreateResultCounter.WithLabelValues(
			strconv.FormatUint(bizChangefeedEnt.ClusterID.Uint64(), 10),
			strconv.FormatUint(bizChangefeedEnt.ClusterID.Uint64(), 10),
			genChangefeedIDStrInMetric(bizChangefeedEnt.ID.Uint64()),
			internal_metrics.NextGen,
			changefeedCreateResultFailed,
			bizChangefeedEnt.ErrMsg,
			errwrapper.ChangefeedErrType(bizChangefeedEnt.ErrMsg)).
			Inc()
	}
}

func convertChangefeedStatus(
	bizChangefeedEnt *entity.BizChangefeed, kernelChangefeedEnt *entity.TiCDCChangefeed,
) entity.CDCStatus {
	if bizChangefeedEnt.OperationStatus != entity.BizChangefeedOperationStatusNormal {
		return bizChangefeedOperationStatus2CDCStatus(bizChangefeedEnt.OperationStatus)
	} else if kernelChangefeedEnt == nil {
		return entity.CDCStatusNormal
	} else {
		return kernelChangefeedStatus2CDCStatus(kernelChangefeedEnt.KernelMetrics.FeedState)
	}
}

func bizChangefeedOperationStatus2CDCStatus(status entity.BizChangefeedOperationStatus) entity.CDCStatus {
	switch status {
	case entity.BizChangefeedOperationStatusCreating:
		return entity.CDCStatusCreating
	case entity.BizChangefeedOperationStatusNormal:
		return entity.CDCStatusNormal
	case entity.BizChangefeedOperationStatusScaling:
		return entity.CDCStatusScaling
	case entity.BizChangefeedOperationStatusPaused:
		return entity.CDCStatusStopped
	case entity.BizChangefeedOperationStatusDeleting:
		return entity.CDCStatusDeleting
	case entity.BizChangefeedOperationStatusDeleted:
		return entity.CDCStatusDeleted
	case entity.BizChangefeedOperationStatusFailed:
		return entity.CDCStatusFailed
	default:
		return entity.CDCStatusNormal
	}
}

func kernelChangefeedStatus2CDCStatus(status entity.ChangefeedKernelStatus) entity.CDCStatus {
	switch status {
	case entity.ChangefeedKernelStatusNormal:
		return entity.CDCStatusNormal
	case entity.ChangefeedKernelStatusStopped:
		return entity.CDCStatusStopped
	case entity.ChangefeedKernelStatusWarning, entity.ChangefeedKernelStatusError:
		return entity.CDCStatusError
	case entity.ChangefeedKernelStatusFailed:
		return entity.CDCStatusFailed
	default:
		return entity.CDCStatusNormal
	}
}

func genChangefeedIDStrInMetric(bizChangefeedID uint64) string {
	return fmt.Sprintf("changefeed-%d", bizChangefeedID)
}