package datapipeline

import (
	"context"
	"strconv"
	"strings"
	"time"

	"github.com/pingcap/log"
	"github.com/prometheus/client_golang/prometheus"
	data_pipeline_repo "github.com/tidbcloud/dataflow-service-ng/app/repo/data_pipeline"
	"github.com/tidbcloud/dataflow-service-ng/app/repo/entity"
	"go.uber.org/zap"
)

const (
	// recentFailWindow matches the premium DM collector. Long enough that a
	// scrape gap does not drop a failure on the floor, short enough that the
	// alert reads as "this just broke" rather than "this is still broken",
	// which is what StatusCount is for.
	recentFailWindow = 15 * time.Minute

	// collectPageSize matches the observer's default so that a scrape reads the
	// table the same way the writer walks it.
	collectPageSize = 100

	unknownErrorCode = "unknown"
)

// collectedStatuses is every status except Completed. Completed pipelines are
// terminal and accumulate without bound, and no alert would ever look at them.
var collectedStatuses = []entity.DataPipelineStatus{
	entity.DataPipelineStatusCreating,
	entity.DataPipelineStatusRunning,
	entity.DataPipelineStatusPausing,
	entity.DataPipelineStatusPaused,
	entity.DataPipelineStatusEditing,
	entity.DataPipelineStatusResuming,
	entity.DataPipelineStatusDeleting,
	entity.DataPipelineStatusFailed,
}

// observedStatuses are the statuses the observer polls. Reporting staleness for
// anything else would alert on pipelines that nothing is observing by design.
var observedStatuses = map[entity.DataPipelineStatus]struct{}{
	entity.DataPipelineStatusCreating: {},
	entity.DataPipelineStatusRunning:  {},
	entity.DataPipelineStatusPausing:  {},
	entity.DataPipelineStatusPaused:   {},
	entity.DataPipelineStatusResuming: {},
	entity.DataPipelineStatusFailed:   {},
}

var _ prometheus.Collector = (*Collector)(nil)

// Collector reports every data pipeline signal from a single pass over the
// table. The changefeed metrics are pushed from a worker and need a companion
// job to delete the series of rows that have gone away; reading at scrape time
// has no such problem, because a row that is no longer listed simply stops
// being reported.
type Collector struct {
	repo data_pipeline_repo.DataPipelineRepo
}

func NewCollector(repo data_pipeline_repo.DataPipelineRepo) prometheus.Collector {
	return &Collector{repo: repo}
}

func (c *Collector) Describe(ch chan<- *prometheus.Desc) {
	ch <- StatusCount
	ch <- StatusDurationSeconds
	ch <- RecentFail
	ch <- ObservationStalenessSeconds
	ch <- LatencySeconds
	ch <- ChildError
}

func (c *Collector) Collect(ch chan<- prometheus.Metric) {
	if err := c.collect(context.Background(), ch); err != nil {
		log.L().Error("failed to collect data pipeline metrics", zap.Error(err))
	}
}

func (c *Collector) collect(ctx context.Context, ch chan<- prometheus.Metric) error {
	if c.repo == nil {
		return nil
	}
	pipelines, err := c.list(ctx)
	if err != nil {
		return err
	}
	c.collectFromPipelines(pipelines, time.Now().UTC(), ch)
	return nil
}

func (c *Collector) list(ctx context.Context) ([]entity.DataPipeline, error) {
	var all []entity.DataPipeline
	lastID := uint64(0)
	for {
		page, err := c.repo.List(
			ctx,
			data_pipeline_repo.WithIDAfter(lastID),
			data_pipeline_repo.WithStatuses(collectedStatuses...),
			data_pipeline_repo.OrderByIDAsc(),
			data_pipeline_repo.WithPagination(0, collectPageSize),
		)
		if err != nil {
			return nil, err
		}
		all = append(all, page...)
		if len(page) < collectPageSize {
			return all, nil
		}
		lastID = page[len(page)-1].ID.Uint64()
	}
}

func (c *Collector) collectFromPipelines(
	pipelines []entity.DataPipeline,
	now time.Time,
	ch chan<- prometheus.Metric,
) {
	statusCount := make(map[statusKey]float64, len(pipelines))
	for index := range pipelines {
		pipeline := &pipelines[index]
		tenant := strconv.FormatUint(pipeline.TenantID.Uint64(), 10)
		cluster := strconv.FormatUint(pipeline.ClusterID.Uint64(), 10)
		id := strconv.FormatUint(pipeline.ID.Uint64(), 10)
		status := string(pipeline.Status)

		statusCount[statusKey{tenant: tenant, cluster: cluster, status: status}]++

		ch <- prometheus.MustNewConstMetric(
			StatusDurationSeconds, prometheus.GaugeValue,
			now.Sub(statusSince(pipeline)).Seconds(),
			tenant, cluster, id, status,
		)

		emitRecentFail(pipeline, now, tenant, cluster, id, ch)
		emitStaleness(pipeline, now, tenant, cluster, id, status, ch)
		emitLatency(pipeline, tenant, cluster, id, ch)
		emitChildErrors(pipeline, tenant, cluster, id, ch)
	}

	for key, count := range statusCount {
		ch <- prometheus.MustNewConstMetric(
			StatusCount, prometheus.GaugeValue, count,
			key.tenant, key.cluster, key.status,
		)
	}
}

type statusKey struct {
	tenant  string
	cluster string
	status  string
}

// statusSince reports when the pipeline entered its current status.
//
// It must not be derived from UpdatedAt. The observation worker saves every
// polled row on every pass - creating, running, pausing, paused, resuming and
// failed all get written every interval - so UpdatedAt says "when we last
// looked", which for a stuck pipeline is always a few seconds ago. A duration
// built on it would sit near zero forever and the stuck-lifecycle alerts would
// never fire, which is worse than having no metric at all: the dashboard would
// look healthy.
//
// StatusEnteredAt is written only by the status setter. Rows predating it fall
// back to creation, which over-reports the age rather than under-reporting it.
func statusSince(pipeline *entity.DataPipeline) time.Time {
	if pipeline.StatusEnteredAt != nil {
		return *pipeline.StatusEnteredAt
	}
	return pipeline.CreatedAt
}

func emitRecentFail(
	pipeline *entity.DataPipeline,
	now time.Time,
	tenant, cluster, id string,
	ch chan<- prometheus.Metric,
) {
	if pipeline.Status != entity.DataPipelineStatusFailed {
		return
	}
	failedAt := statusSince(pipeline)
	code := unknownErrorCode
	if pipeline.LastError != nil {
		if pipeline.LastError.OccurredAt != nil {
			failedAt = *pipeline.LastError.OccurredAt
		}
		if pipeline.LastError.Code != "" {
			code = pipeline.LastError.Code
		}
	}
	elapsed := now.Sub(failedAt)
	if elapsed < 0 || elapsed > recentFailWindow {
		return
	}
	ch <- prometheus.MustNewConstMetric(
		RecentFail, prometheus.GaugeValue, elapsed.Seconds(),
		tenant, cluster, id, code,
	)
}

func emitStaleness(
	pipeline *entity.DataPipeline,
	now time.Time,
	tenant, cluster, id, status string,
	ch chan<- prometheus.Metric,
) {
	if _, observed := observedStatuses[pipeline.Status]; !observed {
		return
	}
	// A pipeline that has never been observed successfully is measured from
	// creation. Reporting nothing would hide the worst case: an observer that
	// has not managed a single successful poll since the row appeared.
	since := pipeline.CreatedAt
	if at := pipeline.Observation.LastSucceededAt; at != nil {
		since = *at
	}
	staleness := now.Sub(since).Seconds()
	if staleness < 0 {
		staleness = 0
	}
	ch <- prometheus.MustNewConstMetric(
		ObservationStalenessSeconds, prometheus.GaugeValue, staleness,
		tenant, cluster, id, status,
	)
}

func emitLatency(
	pipeline *entity.DataPipeline,
	tenant, cluster, id string,
	ch chan<- prometheus.Metric,
) {
	observation := pipeline.Observation
	if observation.ChangefeedCheckpointTSO == 0 || observation.LakeResolvedTSO == 0 {
		return
	}
	ch <- prometheus.MustNewConstMetric(
		LatencySeconds, prometheus.GaugeValue, float64(observation.LatencySeconds),
		tenant, cluster, id,
	)
}

func emitChildErrors(
	pipeline *entity.DataPipeline,
	tenant, cluster, id string,
	ch chan<- prometheus.Metric,
) {
	// Only while the pipeline is supposed to be moving. A paused pipeline's
	// children are stopped on purpose, and a deleting one is being torn down.
	if !isLiveStatus(pipeline.Status) {
		return
	}
	observation := pipeline.Observation
	for child, message := range map[string]string{
		"export":     observation.ExportErrorMessage,
		"changefeed": observation.ChangefeedErrorMessage,
		"lake":       observation.LakeErrorMessage,
	} {
		if strings.TrimSpace(message) == "" {
			continue
		}
		ch <- prometheus.MustNewConstMetric(
			ChildError, prometheus.GaugeValue, 1,
			tenant, cluster, id, child,
		)
	}
}

// isLiveStatus mirrors the observer's notion of a pipeline that is expected to
// be delivering rows.
func isLiveStatus(status entity.DataPipelineStatus) bool {
	return status == entity.DataPipelineStatusCreating ||
		status == entity.DataPipelineStatusRunning ||
		status == entity.DataPipelineStatusResuming
}