package metric

import (
	"context"
	"time"

	"github.com/prometheus/client_golang/prometheus"
	"github.com/tidbcloud/pkgv2/log"
	"github.com/tidbcloud/pkgv2/tracing"
	"github.com/tidbcloud/serverless-service/internal/domain/constant"
	"github.com/tidbcloud/serverless-service/internal/domain/entity"
	"github.com/tidbcloud/serverless-service/internal/domain/repo"
	"github.com/tidbcloud/serverless-service/internal/utils"
	"go.uber.org/zap"
)

const (
	CDCSubsystem = "cdc"

	maxChangefeedCreatingTime = 10 * time.Minute
	maxChangefeedDelayTime    = 30 * time.Minute
	recentFailDuration        = 30 * time.Minute
)

var (
	ChangefeedCount = prometheus.NewDesc(
		prometheus.BuildFQName(MetricNamespace, CDCSubsystem, "count"),
		"Count of changefeeds",
		[]string{"state"},
		nil)

	// metric used for alert
	ChangefeedState = prometheus.NewDesc(
		prometheus.BuildFQName(MetricNamespace, CDCSubsystem, "state"),
		"Failed or warning state of changefeed",
		[]string{"cluster_id", "changefeed_id", "state", "org_id"},
		nil)

	ChangefeedStateMisMatchCounter = prometheus.NewCounterVec(
		prometheus.CounterOpts{
			Namespace: MetricNamespace,
			Subsystem: CDCSubsystem,
			Name:      "state_mismatch_total",
			Help:      "Counter for CDC state mismatch",
		},
		[]string{"cluster_id", "changefeed_id", "db_state", "changefeed_state"},
	)

	ChangefeedCheckPointDealy = prometheus.NewDesc(
		prometheus.BuildFQName(MetricNamespace, CDCSubsystem, "checkpoint_delay_seconds"),
		"Delay checkpoint of changefeed",
		[]string{"cluster_id", "changefeed_id", "org_id"},
		nil)

	ChangefeedCreatingTooLong = prometheus.NewDesc(
		prometheus.BuildFQName(MetricNamespace, CDCSubsystem, "creating_duration_seconds"),
		"Duration in seconds to creating changefeed",
		[]string{"cluster_id", "changefeed_id", "org_id"},
		nil)
)

func init() {
	prometheus.MustRegister(ChangefeedStateMisMatchCounter)
}

type ChangefeedMetricCollector struct {
	changfeedRepo repo.ChangefeedRepo
}

func NewChangefeedMetricCollector(changfeedRepo repo.ChangefeedRepo) *ChangefeedMetricCollector {
	return &ChangefeedMetricCollector{
		changfeedRepo: changfeedRepo,
	}
}

func (c *ChangefeedMetricCollector) Describe(ch chan<- *prometheus.Desc) {
	ch <- ChangefeedCount
	ch <- ChangefeedCheckPointDealy
	ch <- ChangefeedCreatingTooLong
}

func (c *ChangefeedMetricCollector) Collect(ch chan<- prometheus.Metric) {
	ctx := context.Background()
	var err error
	ctx, span := ExportTracer.Start(ctx, "MetricsCollector.Collect")
	defer func() {
		tracing.EndSpan(span, err)
	}()

	log.Info(ctx, "start to collect ChangefeedState and ChangefeedCreatingTooLong metric")
	recentFailTime := time.Now().Add(-recentFailDuration)
	changefeeds, err := c.changfeedRepo.ListByFilter(ctx, entity.ListChangefeedsFilter{
		StatesIn: []constant.ChangfeedState{
			constant.ChangefeedStateFailed,
			constant.ChangefeedStateWarning,
			constant.ChangefeedStateCreating,
			constant.ChangefeedStateCreateFailed,
		},
		FailedTimeGTE: &recentFailTime,
	})
	if err != nil {
		log.Error(ctx, "failed to list changefeeds", zap.Error(err))
	} else {
		for _, cf := range changefeeds {
			switch cf.State {
			case constant.ChangefeedStateFailed, constant.ChangefeedStateWarning, constant.ChangefeedStateCreateFailed:
				// This alert metric intentionally tracks only recent failures to avoid
				// permanently alerting on historical failed/warning records.
				ch <- prometheus.MustNewConstMetric(
					ChangefeedState,
					prometheus.GaugeValue,
					1,
					cf.ClusterID,
					cf.ID,
					string(cf.State),
					cf.TenantID,
				)
			case constant.ChangefeedStateCreating:
				duration := time.Since(cf.CreatedAt)
				if duration > maxChangefeedCreatingTime {
					ch <- prometheus.MustNewConstMetric(
						ChangefeedCreatingTooLong,
						prometheus.GaugeValue,
						duration.Seconds(),
						cf.ClusterID,
						cf.ID,
						cf.TenantID,
					)
					log.Warn(ctx, "changefeed creating too long", zap.String("id", cf.ID))
				}
			}
		}
	}

	log.Info(ctx, "start to collect ChangefeedCheckPointDealy metric")
	now := time.Now()
	delayChangefeeds, err := c.changfeedRepo.ListByFilter(ctx, entity.ListChangefeedsFilter{
		CheckpointTSOLT: utils.TimeToTSO(now.Add(-maxChangefeedDelayTime)),
		State:           constant.ChangfeedStateRunning,
	})
	if err != nil {
		log.Error(ctx, "failed to list check point delay changefeeds", zap.Error(err))
	} else {
		for _, cf := range delayChangefeeds {
			// A newly created changefeed has no checkpoint yet, its zero TSO would be
			// converted to the unix epoch and reported as a decades-long delay.
			if !utils.IsValidTSO(cf.CheckpointTSO) {
				log.Info(ctx, "changefeed checkpoint not initialized yet, skip delay metric",
					zap.String("cluster_id", cf.ClusterID), zap.String("changefeed_id", cf.ID))
				continue
			}
			checkPointTime := utils.TSOToTime(cf.CheckpointTSO)
			delaySeconds := now.Sub(checkPointTime).Seconds()
			ch <- prometheus.MustNewConstMetric(
				ChangefeedCheckPointDealy,
				prometheus.GaugeValue,
				delaySeconds,
				cf.ClusterID,
				cf.ID,
				cf.TenantID,
			)
		}
	}

	log.Info(ctx, "start to collect ChangefeedCount metric")
	stateCount, err := c.changfeedRepo.GroupByState(ctx)
	if err != nil {
		log.Error(ctx, "failed to count changfeed", zap.Error(err))
	} else {
		for _, sc := range stateCount {
			ch <- prometheus.MustNewConstMetric(
				ChangefeedCount,
				prometheus.GaugeValue,
				float64(sc.Count),
				string(sc.State),
			)
		}
	}
}