package datapipeline

import (
	"context"
	"sort"
	"strings"
	"testing"
	"time"

	"github.com/prometheus/client_golang/prometheus"
	io_prometheus_client "github.com/prometheus/client_model/go"
	"github.com/stretchr/testify/require"
	data_pipeline_repo "github.com/tidbcloud/dataflow-service-ng/app/repo/data_pipeline"
	"github.com/tidbcloud/dataflow-service-ng/app/repo/entity"
	"github.com/tidbcloud/pkgv2/bizerr"
)

// sample is one collected metric flattened into something a test can assert on
// without caring about the order the collector emitted things in.
type sample struct {
	name   string
	labels map[string]string
	value  float64
}

func drain(t *testing.T, ch chan prometheus.Metric) []sample {
	t.Helper()
	close(ch)
	samples := make([]sample, 0, len(ch))
	for metric := range ch {
		written := &io_prometheus_client.Metric{}
		require.NoError(t, metric.Write(written))
		labels := make(map[string]string, len(written.GetLabel()))
		for _, pair := range written.GetLabel() {
			labels[pair.GetName()] = pair.GetValue()
		}
		// Desc.String() is the only handle on the metric name from this side of
		// the channel; the fqName appears as `fqName: "..."`.
		desc := metric.Desc().String()
		name := desc[strings.Index(desc, `fqName: "`)+len(`fqName: "`):]
		name = name[:strings.Index(name, `"`)]
		samples = append(samples, sample{name: name, labels: labels, value: written.GetGauge().GetValue()})
	}
	return samples
}

func find(samples []sample, name string, match map[string]string) []sample {
	found := make([]sample, 0, 1)
sampleLoop:
	for _, s := range samples {
		if s.name != name {
			continue
		}
		for key, want := range match {
			if s.labels[key] != want {
				continue sampleLoop
			}
		}
		found = append(found, s)
	}
	return found
}

func pipeline(id uint64, status entity.DataPipelineStatus, age time.Duration, now time.Time) entity.DataPipeline {
	enteredAt := now.Add(-age)
	// UpdatedAt is deliberately recent on every fixture: the observation worker
	// rewrites every polled row each interval, so a collector that read it
	// would report a few seconds for a pipeline that has been stuck for hours.
	justSaved := now.Add(-5 * time.Second)
	return entity.DataPipeline{
		ID:              entity.ID(id),
		TenantID:        entity.ID(1001),
		ClusterID:       entity.ID(2001),
		Status:          status,
		CreatedAt:       now.Add(-age - time.Hour),
		StatusEnteredAt: &enteredAt,
		UpdatedAt:       &justSaved,
	}
}

func TestStatusCountGroupsByStatusAndSkipsNothingItWasGiven(t *testing.T) {
	now := time.Now().UTC()
	pipelines := []entity.DataPipeline{
		pipeline(1, entity.DataPipelineStatusRunning, time.Minute, now),
		pipeline(2, entity.DataPipelineStatusRunning, time.Minute, now),
		pipeline(3, entity.DataPipelineStatusFailed, time.Minute, now),
	}

	ch := make(chan prometheus.Metric, 64)
	(&Collector{}).collectFromPipelines(pipelines, now, ch)
	samples := drain(t, ch)

	running := find(samples, "dataflow_data_pipeline_status_count", map[string]string{"status": "running"})
	require.Len(t, running, 1)
	require.Equal(t, 2.0, running[0].value)
	require.Equal(t, "1001", running[0].labels["tenant"])
	require.Equal(t, "2001", running[0].labels["cluster_id"])

	failed := find(samples, "dataflow_data_pipeline_status_count", map[string]string{"status": "failed"})
	require.Len(t, failed, 1)
	require.Equal(t, 1.0, failed[0].value)
}

// The regression this pins: UpdatedAt moves every observation interval, so a
// duration derived from it stays near zero and the stuck-lifecycle alerts never
// fire. Only StatusEnteredAt answers "how long has it been like this?".
func TestStatusDurationSurvivesRepeatedObservationSaves(t *testing.T) {
	now := time.Now().UTC()
	stuck := pipeline(1, entity.DataPipelineStatusCreating, 90*time.Minute, now)
	justSaved := now.Add(-time.Second)
	stuck.UpdatedAt = &justSaved

	ch := make(chan prometheus.Metric, 64)
	(&Collector{}).collectFromPipelines([]entity.DataPipeline{stuck}, now, ch)
	samples := drain(t, ch)

	duration := find(samples, "dataflow_data_pipeline_status_duration_seconds", map[string]string{"pipeline_id": "1"})
	require.Len(t, duration, 1)
	require.InDelta(t, (90 * time.Minute).Seconds(), duration[0].value, 1)
}

// A row written before StatusEnteredAt existed measures from creation, which
// over-reports the age rather than hiding it.
func TestStatusDurationFallsBackToCreation(t *testing.T) {
	now := time.Now().UTC()
	legacy := pipeline(1, entity.DataPipelineStatusCreating, time.Minute, now)
	legacy.StatusEnteredAt = nil
	legacy.CreatedAt = now.Add(-3 * time.Hour)

	ch := make(chan prometheus.Metric, 64)
	(&Collector{}).collectFromPipelines([]entity.DataPipeline{legacy}, now, ch)
	samples := drain(t, ch)

	duration := find(samples, "dataflow_data_pipeline_status_duration_seconds", map[string]string{"pipeline_id": "1"})
	require.Len(t, duration, 1)
	require.InDelta(t, (3 * time.Hour).Seconds(), duration[0].value, 1)
}

func TestStatusDurationMeasuresTimeInCurrentStatus(t *testing.T) {
	now := time.Now().UTC()
	pipelines := []entity.DataPipeline{pipeline(1, entity.DataPipelineStatusCreating, 40*time.Minute, now)}

	ch := make(chan prometheus.Metric, 64)
	(&Collector{}).collectFromPipelines(pipelines, now, ch)
	samples := drain(t, ch)

	duration := find(samples, "dataflow_data_pipeline_status_duration_seconds", map[string]string{"pipeline_id": "1"})
	require.Len(t, duration, 1)
	require.InDelta(t, (40 * time.Minute).Seconds(), duration[0].value, 1)
	require.Equal(t, "creating", duration[0].labels["status"])
}

func TestRecentFailIsAnEventNotALevel(t *testing.T) {
	now := time.Now().UTC()
	justFailedAt := now.Add(-2 * time.Minute)
	longFailedAt := now.Add(-3 * time.Hour)

	fresh := pipeline(1, entity.DataPipelineStatusFailed, 2*time.Minute, now)
	fresh.LastError = &entity.DataPipelineError{Code: "CHILD_FAILED", OccurredAt: &justFailedAt}
	stale := pipeline(2, entity.DataPipelineStatusFailed, 3*time.Hour, now)
	stale.LastError = &entity.DataPipelineError{Code: "CHILD_FAILED", OccurredAt: &longFailedAt}

	ch := make(chan prometheus.Metric, 64)
	(&Collector{}).collectFromPipelines([]entity.DataPipeline{fresh, stale}, now, ch)
	samples := drain(t, ch)

	require.Len(t, find(samples, "dataflow_data_pipeline_recent_fail", map[string]string{"pipeline_id": "1"}), 1)
	require.Equal(t, "CHILD_FAILED",
		find(samples, "dataflow_data_pipeline_recent_fail", map[string]string{"pipeline_id": "1"})[0].labels["error_code"])

	// The old failure has dropped out of the window, but it is still counted as
	// a failed pipeline - that is what StatusCount is for.
	require.Empty(t, find(samples, "dataflow_data_pipeline_recent_fail", map[string]string{"pipeline_id": "2"}))
	require.Equal(t, 2.0,
		find(samples, "dataflow_data_pipeline_status_count", map[string]string{"status": "failed"})[0].value)
}

func TestRecentFailFallsBackToTheRowTimestampAndAnUnknownCode(t *testing.T) {
	now := time.Now().UTC()
	failed := pipeline(1, entity.DataPipelineStatusFailed, 3*time.Minute, now)
	failed.LastError = nil

	ch := make(chan prometheus.Metric, 64)
	(&Collector{}).collectFromPipelines([]entity.DataPipeline{failed}, now, ch)
	samples := drain(t, ch)

	found := find(samples, "dataflow_data_pipeline_recent_fail", map[string]string{"pipeline_id": "1"})
	require.Len(t, found, 1)
	require.Equal(t, unknownErrorCode, found[0].labels["error_code"])
	require.InDelta(t, (3 * time.Minute).Seconds(), found[0].value, 1)
}

func TestStalenessMeasuresFromTheLastSuccessfulObservation(t *testing.T) {
	now := time.Now().UTC()
	succeededAt := now.Add(-9 * time.Minute)

	observed := pipeline(1, entity.DataPipelineStatusRunning, time.Minute, now)
	observed.Observation.LastSucceededAt = &succeededAt

	ch := make(chan prometheus.Metric, 64)
	(&Collector{}).collectFromPipelines([]entity.DataPipeline{observed}, now, ch)
	samples := drain(t, ch)

	found := find(samples, "dataflow_data_pipeline_observation_staleness_seconds", map[string]string{"pipeline_id": "1"})
	require.Len(t, found, 1)
	require.InDelta(t, (9 * time.Minute).Seconds(), found[0].value, 1)
	require.Equal(t, "running", found[0].labels["status"])
}

func TestStalenessMeasuresFromCreationWhenNothingHasEverSucceeded(t *testing.T) {
	now := time.Now().UTC()
	// Created two hours ago, never once observed: the worst case, and the one
	// that must not be silently absent.
	never := pipeline(1, entity.DataPipelineStatusRunning, time.Minute, now)
	never.CreatedAt = now.Add(-2 * time.Hour)
	never.Observation.LastSucceededAt = nil

	ch := make(chan prometheus.Metric, 64)
	(&Collector{}).collectFromPipelines([]entity.DataPipeline{never}, now, ch)
	samples := drain(t, ch)

	found := find(samples, "dataflow_data_pipeline_observation_staleness_seconds", map[string]string{"pipeline_id": "1"})
	require.Len(t, found, 1)
	require.InDelta(t, (2 * time.Hour).Seconds(), found[0].value, 1)
}

func TestStalenessIsNotReportedForStatusesNobodyObserves(t *testing.T) {
	now := time.Now().UTC()
	for _, status := range []entity.DataPipelineStatus{
		entity.DataPipelineStatusEditing,
		entity.DataPipelineStatusDeleting,
	} {
		ch := make(chan prometheus.Metric, 64)
		(&Collector{}).collectFromPipelines([]entity.DataPipeline{pipeline(1, status, time.Hour, now)}, now, ch)
		samples := drain(t, ch)
		require.Empty(t,
			find(samples, "dataflow_data_pipeline_observation_staleness_seconds", nil),
			"status %s is not polled by the observer, so staleness would alert on nothing", status)
	}
}

func TestLatencyIsAbsentRatherThanZeroBeforeBothPositionsAreKnown(t *testing.T) {
	now := time.Now().UTC()

	unknown := pipeline(1, entity.DataPipelineStatusRunning, time.Minute, now)
	unknown.Observation.ChangefeedCheckpointTSO = 450000000000000000
	unknown.Observation.LakeResolvedTSO = 0
	unknown.Observation.LatencySeconds = 0

	known := pipeline(2, entity.DataPipelineStatusRunning, time.Minute, now)
	known.Observation.ChangefeedCheckpointTSO = 450000000000000000
	known.Observation.LakeResolvedTSO = 449000000000000000
	known.Observation.LatencySeconds = 42

	ch := make(chan prometheus.Metric, 64)
	(&Collector{}).collectFromPipelines([]entity.DataPipeline{unknown, known}, now, ch)
	samples := drain(t, ch)

	require.Empty(t, find(samples, "dataflow_data_pipeline_latency_seconds", map[string]string{"pipeline_id": "1"}))
	found := find(samples, "dataflow_data_pipeline_latency_seconds", map[string]string{"pipeline_id": "2"})
	require.Len(t, found, 1)
	require.Equal(t, 42.0, found[0].value)
}

func TestChildErrorsAreReportedPerChildWhileThePipelineLooksAlive(t *testing.T) {
	now := time.Now().UTC()
	broken := pipeline(1, entity.DataPipelineStatusRunning, time.Minute, now)
	broken.Observation.ExportErrorMessage = "export failed"
	broken.Observation.LakeErrorMessage = "Lake integration run failed"
	broken.Observation.ChangefeedErrorMessage = "   " // whitespace is not an error

	ch := make(chan prometheus.Metric, 64)
	(&Collector{}).collectFromPipelines([]entity.DataPipeline{broken}, now, ch)
	samples := drain(t, ch)

	children := make([]string, 0, 2)
	for _, s := range find(samples, "dataflow_data_pipeline_child_error", nil) {
		require.Equal(t, 1.0, s.value)
		children = append(children, s.labels["child"])
	}
	sort.Strings(children)
	require.Equal(t, []string{"export", "lake"}, children)
}

func TestChildErrorsAreSilentOnceThePipelineIsMeantToBeStopped(t *testing.T) {
	now := time.Now().UTC()
	for _, status := range []entity.DataPipelineStatus{
		entity.DataPipelineStatusPaused,
		entity.DataPipelineStatusDeleting,
	} {
		stopped := pipeline(1, status, time.Minute, now)
		stopped.Observation.ChangefeedErrorMessage = "changefeed is stopped"

		ch := make(chan prometheus.Metric, 64)
		(&Collector{}).collectFromPipelines([]entity.DataPipeline{stopped}, now, ch)
		samples := drain(t, ch)
		require.Empty(t, find(samples, "dataflow_data_pipeline_child_error", nil),
			"a %s pipeline's children are stopped on purpose", status)
	}
}

func TestListPagesUntilTheTableIsExhausted(t *testing.T) {
	for name, total := range map[string]int{
		"partial final page": collectPageSize + 7,
		// The dangerous size: a final page that is exactly full looks like
		// "there may be more", so the loop must go round once more and get
		// nothing rather than deciding it is done or running forever.
		"exactly one full page": collectPageSize,
		"empty table":           0,
	} {
		t.Run(name, func(t *testing.T) {
			repo := &pagingRepo{total: total}
			collected, err := (&Collector{repo: repo}).list(context.Background())
			require.NoError(t, err)
			require.Len(t, collected, total)
			require.Equal(t, total/collectPageSize+1, repo.calls)
			// Every row appears once: a cursor that failed to advance would
			// repeat the first page instead.
			seen := make(map[entity.ID]bool, len(collected))
			for _, p := range collected {
				require.False(t, seen[p.ID], "pipeline %d collected twice", p.ID)
				seen[p.ID] = true
			}
		})
	}
}

// pagingRepo serves a synthetic table one page at a time. It advances its own
// cursor rather than reading the collector's, because a FilterOp is an opaque
// closure over a query set that cannot be inspected without a database; what
// the test can still prove is that the loop terminates and never repeats a row.
type pagingRepo struct {
	data_pipeline_repo.DataPipelineRepo
	total  int
	calls  int
	served int
}

func (r *pagingRepo) List(
	_ context.Context,
	_ ...data_pipeline_repo.FilterOp,
) ([]entity.DataPipeline, bizerr.BizError) {
	r.calls++
	page := make([]entity.DataPipeline, 0, collectPageSize)
	for r.served < r.total && len(page) < collectPageSize {
		r.served++
		page = append(page, entity.DataPipeline{ID: entity.ID(r.served)})
	}
	return page, nil
}